putobject reduce_read_quorum_errs

.docker/Dockerfile.devenv

@@ -0,0 +1,29 @@
+FROM m.daocloud.io/docker.io/library/ubuntu:22.04
+
+ENV LANG C.UTF-8
+
+RUN sed -i s@http://.*archive.ubuntu.com@http://repo.huaweicloud.com@g /etc/apt/sources.list
+
+RUN apt-get clean && apt-get update && apt-get install wget git curl unzip gcc pkg-config libssl-dev -y
+
+# install protoc
+RUN wget https://github.com/protocolbuffers/protobuf/releases/download/v27.0/protoc-27.0-linux-x86_64.zip \
+    && unzip protoc-27.0-linux-x86_64.zip -d protoc3 \
+    && mv protoc3/bin/* /usr/local/bin/ && chmod +x /usr/local/bin/protoc && mv protoc3/include/* /usr/local/include/ && rm -rf protoc-27.0-linux-x86_64.zip protoc3
+
+# install flatc
+RUN wget https://github.com/google/flatbuffers/releases/download/v24.3.25/Linux.flatc.binary.g++-13.zip \
+    && unzip Linux.flatc.binary.g++-13.zip \
+    && mv flatc /usr/local/bin/ && chmod +x /usr/local/bin/flatc && rm -rf Linux.flatc.binary.g++-13.zip
+
+# install rust
+ENV RUSTUP_DIST_SERVER="https://rsproxy.cn"
+ENV RUSTUP_UPDATE_ROOT="https://rsproxy.cn/rustup"
+RUN curl -o rustup-init.sh --proto '=https' --tlsv1.2 -sSf https://rsproxy.cn/rustup-init.sh \
+    && sh rustup-init.sh -y && rm -rf rustup-init.sh
+
+COPY .docker/cargo.config.toml /root/.cargo/config.toml
+
+WORKDIR /root/s3-rustfs
+
+CMD [ "bash", "-c", "while true; do sleep 1; done" ]

.docker/cargo.config.toml

@@ -0,0 +1,13 @@
+[source.crates-io]
+registry = "https://github.com/rust-lang/crates.io-index"
+replace-with = 'rsproxy-sparse'
+
+[source.rsproxy]
+registry = "https://rsproxy.cn/crates.io-index"
+[registries.rsproxy]
+index = "https://rsproxy.cn/crates.io-index"
+[source.rsproxy-sparse]
+registry = "sparse+https://rsproxy.cn/index/"
+
+[net]
+git-fetch-with-cli = true

.github/workflows/rust.yml

@@ -1,8 +1,8 @@
 name: Rust
 
 on:
+  workflow_dispatch:
   push:
-    branches: [ "main" ]
   pull_request:
     branches: [ "main" ]
 
@@ -11,12 +11,68 @@ env:
 
 jobs:
   build:
-
     runs-on: ubuntu-latest
+    strategy:
+      matrix:
+        rust:
+          - stable
+          - beta
+          - nightly
 
     steps:
-    - uses: actions/checkout@v4
+      - name: cache protoc bin
-    - name: Build
+        id: cache-protoc-action
-      run: cargo build --verbose
+        uses: actions/cache@v3
-    - name: Run tests
+        env:
-      run: cargo test --verbose
+          cache-name: cache-protoc-action-bin
+        with:
+          path: /usr/local/bin/protoc
+          key: ${{ runner.os }}-build-${{ env.cache-name }}-v0.0.1
+
+      - name: install protoc
+        if: steps.cache-protoc-action.outputs.cache-hit != 'true'
+        run: |
+          wget https://github.com/protocolbuffers/protobuf/releases/download/v27.0/protoc-27.0-linux-x86_64.zip
+          unzip protoc-27.0-linux-x86_64.zip -d protoc3
+          mv protoc3/bin/* /usr/local/bin/
+          chmod +x /usr/local/bin/protoc
+          rm -rf protoc-27.0-linux-x86_64.zip protoc3
+
+      - name: print protoc version
+        run: protoc --version
+
+      - name: cache flatc bin
+        id: cache-flatc-action
+        uses: actions/cache@v3
+        env:
+          cache-name: cache-flatc-action-bin
+        with:
+          path: /usr/local/bin/flatc
+          key: ${{ runner.os }}-build-${{ env.cache-name }}-v0.0.1
+
+      - name: install flatc
+        if: steps.cache-flatc-action.outputs.cache-hit != 'true'
+        run: |
+          wget https://github.com/google/flatbuffers/releases/download/v24.3.25/Linux.flatc.binary.g++-13.zip
+          unzip Linux.flatc.binary.g++-13.zip
+          mv flatc /usr/local/bin/ 
+          chmod +x /usr/local/bin/flatc 
+          rm -rf Linux.flatc.binary.g++-13.zip
+
+      - uses: actions/checkout@v2
+
+      - uses: actions-rs/toolchain@v1
+        with:
+          profile: minimal
+          toolchain: ${{ matrix.rust }}
+          override: true
+          components: rustfmt, clippy
+
+      - uses: actions-rs/cargo@v1
+        with:
+          command: build
+
+      - uses: actions-rs/cargo@v1
+        with:
+          command: test
+          args: --all

Cargo.lock

@@ -447,6 +447,7 @@ dependencies = [
  "http",
  "lazy_static",
  "netif",
+ "num_cpus",
  "openssl",
  "path-absolutize",
  "path-clean",

Makefile

@@ -0,0 +1,23 @@
+###########
+# 远程开发，需要 VSCode 安装 Dev Containers, Remote SSH, Remote Explorer
+# https://code.visualstudio.com/docs/remote/containers
+###########
+DOCKER_CLI ?= docker
+IMAGE_NAME ?= rustfs:v1.0.0
+CONTAINER_NAME ?= rustfs-dev
+DOCKERFILE ?= $(shell pwd)/.docker/Dockerfile.devenv
+
+.PHONY: init-devenv
+init-devenv:
+	$(DOCKER_CLI) build -t $(IMAGE_NAME) -f $(DOCKERFILE) .
+	$(DOCKER_CLI) stop $(CONTAINER_NAME)
+	$(DOCKER_CLI) rm $(CONTAINER_NAME)
+	$(DOCKER_CLI) run -d --name $(CONTAINER_NAME) -p 9010:9010 -p 9000:9000 -v $(shell pwd):/root/s3-rustfs -it $(IMAGE_NAME)
+
+.PHONY: start
+start:
+	$(DOCKER_CLI) start $(CONTAINER_NAME)
+
+.PHONY: stop
+stop:
+	$(DOCKER_CLI) stop $(CONTAINER_NAME)	

ecstore/Cargo.toml

@@ -38,13 +38,14 @@ base64-simd = "0.8.0"
 sha2 = "0.10.8"
 hex-simd = "0.8.0"
 path-clean = "1.0.1"
-tokio = { workspace = true, features = ["io-util"] }
+tokio = { workspace = true, features = ["io-util", "sync"] }
 tokio-stream = "0.1.15"
 tonic.workspace = true
 tower.workspace = true
 rmp = "0.8.14"
 byteorder = "1.5.0"
 xxhash-rust = { version = "0.8.12", features = ["xxh64"] }
+num_cpus = "1.16"
 
 [target.'cfg(not(windows))'.dependencies]
 openssl = "0.10.66"

ecstore/src/disk/local.rs

@@ -901,7 +901,7 @@ impl DiskAPI for LocalDisk {
             self.rename_all(&src_data_path, &dst_data_path, &skip_parent).await?;
         }
 
-        warn!("old_data_dir {:?}", old_data_dir);
+        // warn!("old_data_dir {:?}", old_data_dir);
         // 有旧目录，把old xl.meta存到旧目录里
         if old_data_dir.is_some() {
             self.write_all(

ecstore/src/sets.rs

@@ -1,4 +1,5 @@
-use std::{collections::HashMap, sync::Arc, time::Duration};
+#![allow(clippy::map_entry)]
+use std::{collections::HashMap, sync::Arc};
 
 use crate::{
     disk::{
@@ -19,11 +20,13 @@ use crate::{
 use futures::future::join_all;
 use http::HeaderMap;
 use tokio::sync::RwLock;
+use tokio::sync::Semaphore;
+use tokio::time::Duration;
 use tokio_util::sync::CancellationToken;
 use tracing::{debug, warn};
 use uuid::Uuid;
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub struct Sets {
     pub id: Uuid,
     // pub sets: Vec<Objects>,
@@ -254,8 +257,7 @@ impl StorageAPI for Sets {
         let mut set_obj_map = HashMap::new();
 
         // hash key
-        let mut i = 0;
+        for (i, obj) in objects.iter().enumerate() {
-        for obj in objects.iter() {
             let idx = self.get_hashed_set_index(obj.object_name.as_str());
 
             if !set_obj_map.contains_key(&idx) {
@@ -267,35 +269,40 @@ impl StorageAPI for Sets {
                         obj: obj.clone(),
                     }],
                 );
-            } else {
+            } else if let Some(val) = set_obj_map.get_mut(&idx) {
-                if let Some(val) = set_obj_map.get_mut(&idx) {
+                val.push(DelObj {
-                    val.push(DelObj {
+                    // set_idx: idx,
-                        // set_idx: idx,
+                    orig_idx: i,
-                        orig_idx: i,
+                    obj: obj.clone(),
-                        obj: obj.clone(),
+                });
-                    });
-                }
             }
-
-            i += 1;
         }
 
-        // TODO: 并发
+        let semaphore = Arc::new(Semaphore::new(num_cpus::get()));
+        let mut jhs = Vec::with_capacity(semaphore.available_permits());
+
         for (k, v) in set_obj_map {
             let disks = self.get_disks(k);
-            let objs: Vec<ObjectToDelete> = v.iter().map(|v| v.obj.clone()).collect();
+            let semaphore = semaphore.clone();
-            let (dobjects, errs) = disks.delete_objects(bucket, objs, opts.clone()).await?;
+            let opts = opts.clone();
+            let bucket = bucket.to_string();
 
-            let mut i = 0;
+            let jh = tokio::spawn(async move {
-            for err in errs {
+                let _permit = semaphore.acquire().await.unwrap();
-                let obj = v.get(i).unwrap();
+                let objs: Vec<ObjectToDelete> = v.iter().map(|v| v.obj.clone()).collect();
+                disks.delete_objects(&bucket, objs, opts).await
+            });
+            jhs.push(jh);
+        }
 
-                del_errs[obj.orig_idx] = err;
+        let mut results = Vec::with_capacity(jhs.len());
+        for jh in jhs {
+            results.push(jh.await?.unwrap());
+        }
 
-                del_objects[obj.orig_idx] = dobjects.get(i).unwrap().clone();
+        for (dobjects, errs) in results {
-
+            del_objects.extend(dobjects);
-                i += 1;
+            del_errs.extend(errs);
-            }
         }
 
         Ok((del_objects, del_errs))

ecstore/src/store.rs

@@ -1,3 +1,4 @@
+#![allow(clippy::map_entry)]
 use crate::{
     bucket_meta::BucketMetadata,
     disk::{error::DiskError, new_disk, DiskOption, DiskStore, WalkDirOptions, BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
@@ -23,6 +24,7 @@ use std::{
     time::Duration,
 };
 use time::OffsetDateTime;
+use tokio::sync::Semaphore;
 use tokio::{fs, sync::RwLock};
 use tracing::{debug, info};
 use uuid::Uuid;
@@ -50,10 +52,11 @@ pub async fn update_erasure_type(setup_type: SetupType) {
     *is_erasure_sd = setup_type == SetupType::ErasureSD;
 }
 
+type TypeLocalDiskSetDrives = Vec<Vec<Vec<Option<DiskStore>>>>;
+
 lazy_static! {
     pub static ref GLOBAL_LOCAL_DISK_MAP: Arc<RwLock<HashMap<String, Option<DiskStore>>>> = Arc::new(RwLock::new(HashMap::new()));
-    pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc<RwLock<Vec<Vec<Vec<Option<DiskStore>>>>>> =
+    pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc<RwLock<TypeLocalDiskSetDrives>> = Arc::new(RwLock::new(Vec::new()));
-        Arc::new(RwLock::new(Vec::new()));
 }
 
 pub async fn find_local_disk(disk_path: &String) -> Option<DiskStore> {
@@ -75,7 +78,7 @@ pub async fn find_local_disk(disk_path: &String) -> Option<DiskStore> {
 
 pub async fn all_local_disk_path() -> Vec<String> {
     let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await;
-    disk_map.keys().map(|v| v.clone()).collect()
+    disk_map.keys().cloned().collect()
 }
 
 pub async fn all_local_disk() -> Vec<DiskStore> {
@@ -155,6 +158,7 @@ pub struct ECStore {
 }
 
 impl ECStore {
+    #[allow(clippy::new_ret_no_self)]
     pub async fn new(_address: String, endpoint_pools: EndpointServerPools) -> Result<()> {
         // let layouts = DisksLayout::try_from(endpoints.as_slice())?;
 
@@ -313,8 +317,8 @@ impl ECStore {
 
                         if entry.is_object() {
                             let fi = entry.to_fileinfo(&opts.bucket)?;
-                            if fi.is_some() {
+                            if let Some(f) = fi {
-                                ress.push(fi.unwrap().to_object_info(&opts.bucket, &entry.name, false));
+                                ress.push(f.to_object_info(&opts.bucket, &entry.name, false));
                             }
                             continue;
                         }
@@ -385,62 +389,68 @@ impl ECStore {
         object: &str,
         opts: &ObjectOptions,
     ) -> Result<(PoolObjInfo, Vec<Error>)> {
-        let mut futures = Vec::new();
+        internal_get_pool_info_existing_with_opts(&self.pools, bucket, object, opts).await
-
-        for pool in self.pools.iter() {
-            futures.push(pool.get_object_info(bucket, object, opts));
-        }
-
-        let results = join_all(futures).await;
-
-        let mut ress = Vec::new();
-
-        let mut i = 0;
-
-        // join_all结果跟输入顺序一致
-        for res in results {
-            let index = i;
-
-            match res {
-                Ok(r) => {
-                    ress.push(PoolObjInfo {
-                        index,
-                        object_info: r,
-                        err: None,
-                    });
-                }
-                Err(e) => {
-                    ress.push(PoolObjInfo {
-                        index,
-                        err: Some(e),
-                        ..Default::default()
-                    });
-                }
-            }
-            i += 1;
-        }
-
-        ress.sort_by(|a, b| {
-            let at = a.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
-            let bt = b.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
-
-            at.cmp(&bt)
-        });
-
-        for res in ress {
-            // check
-            if res.err.is_none() {
-                // TODO: let errs = self.poolsWithObject()
-                return Ok((res, Vec::new()));
-            }
-        }
-
-        let ret = PoolObjInfo::default();
-
-        Ok((ret, Vec::new()))
     }
 }
 
+async fn internal_get_pool_info_existing_with_opts(
+    pools: &[Arc<Sets>],
+    bucket: &str,
+    object: &str,
+    opts: &ObjectOptions,
+) -> Result<(PoolObjInfo, Vec<Error>)> {
+    let mut futures = Vec::new();
+
+    for pool in pools.iter() {
+        futures.push(pool.get_object_info(bucket, object, opts));
+    }
+
+    let results = join_all(futures).await;
+
+    let mut ress = Vec::new();
+
+    // join_all结果跟输入顺序一致
+    for (i, res) in results.into_iter().enumerate() {
+        let index = i;
+
+        match res {
+            Ok(r) => {
+                ress.push(PoolObjInfo {
+                    index,
+                    object_info: r,
+                    err: None,
+                });
+            }
+            Err(e) => {
+                ress.push(PoolObjInfo {
+                    index,
+                    err: Some(e),
+                    ..Default::default()
+                });
+            }
+        }
+    }
+
+    ress.sort_by(|a, b| {
+        let at = a.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
+        let bt = b.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
+
+        at.cmp(&bt)
+    });
+
+    for res in ress {
+        // check
+        if res.err.is_none() {
+            // TODO: let errs = self.poolsWithObject()
+            return Ok((res, Vec::new()));
+        }
+    }
+
+    let ret = PoolObjInfo::default();
+
+    Ok((ret, Vec::new()))
+}
+
 #[derive(Debug, Default)]
 pub struct PoolObjInfo {
     pub index: usize,
@@ -553,21 +563,34 @@ impl StorageAPI for ECStore {
             del_errs.push(None)
         }
 
-        // TODO: limte 限制并发数量
+        let mut jhs = Vec::new();
-        let opt = ObjectOptions::default();
+        let semaphore = Arc::new(Semaphore::new(num_cpus::get()));
-        // 取所有poolObjInfo
+        let pools = Arc::new(self.pools.clone());
-        let mut futures = Vec::new();
-        for obj in objects.iter() {
-            futures.push(self.get_pool_info_existing_with_opts(bucket, &obj.object_name, &opt));
-        }
 
-        let results = join_all(futures).await;
+        for obj in objects.iter() {
+            let (semaphore, pools, bucket, object_name, opt) = (
+                semaphore.clone(),
+                pools.clone(),
+                bucket.to_string(),
+                obj.object_name.to_string(),
+                ObjectOptions::default(),
+            );
+
+            let jh = tokio::spawn(async move {
+                let _permit = semaphore.acquire().await.unwrap();
+                internal_get_pool_info_existing_with_opts(pools.as_ref(), &bucket, &object_name, &opt).await
+            });
+            jhs.push(jh);
+        }
+        let mut results = Vec::new();
+        for jh in jhs {
+            results.push(jh.await.unwrap());
+        }
 
         // 记录pool Index 对应的objects pool_idx -> objects idx
         let mut pool_index_objects = HashMap::new();
 
-        let mut i = 0;
+        for (i, res) in results.into_iter().enumerate() {
-        for res in results {
             match res {
                 Ok((pinfo, _)) => {
                     if pinfo.object_info.delete_marker && opts.version_id.is_empty() {
@@ -595,8 +618,6 @@ impl StorageAPI for ECStore {
                     del_errs[i] = Some(e)
                 }
             }
-
-            i += 1;
         }
 
         if !pool_index_objects.is_empty() {
@@ -609,16 +630,7 @@ impl StorageAPI for ECStore {
 
                 let obj_idxs = vals.unwrap();
                 //  取对应obj,理论上不会none
-                let objs: Vec<ObjectToDelete> = obj_idxs
+                let objs: Vec<ObjectToDelete> = obj_idxs.iter().filter_map(|&idx| objects.get(idx).cloned()).collect();
-                    .iter()
-                    .filter_map(|&idx| {
-                        if let Some(obj) = objects.get(idx) {
-                            Some(obj.clone())
-                        } else {
-                            None
-                        }
-                    })
-                    .collect();
 
                 if objs.is_empty() {
                     continue;
@@ -627,8 +639,7 @@ impl StorageAPI for ECStore {
                 let (pdel_objs, perrs) = sets.delete_objects(bucket, objs, opts.clone()).await?;
 
                 // perrs的顺序理论上跟obj_idxs顺序一致
-                let mut i = 0;
+                for (i, err) in perrs.into_iter().enumerate() {
-                for err in perrs {
                     let obj_idx = obj_idxs[i];
 
                     if err.is_some() {
@@ -639,8 +650,6 @@ impl StorageAPI for ECStore {
                     dobj.object_name = utils::path::decode_dir_object(&dobj.object_name);
 
                     del_objects[obj_idx] = dobj;
-
-                    i += 1;
                 }
             }
         }
@@ -649,7 +658,7 @@ impl StorageAPI for ECStore {
     }
     async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result<ObjectInfo> {
         if opts.delete_prefix {
-            self.delete_prefix(bucket, &object).await?;
+            self.delete_prefix(bucket, object).await?;
             return Ok(ObjectInfo::default());
         }