diff --git a/.docker/Dockerfile.devenv b/.docker/Dockerfile.devenv new file mode 100644 index 000000000..e95027d2e --- /dev/null +++ b/.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" ] diff --git a/.docker/cargo.config.toml b/.docker/cargo.config.toml new file mode 100644 index 000000000..ef2fa863f --- /dev/null +++ b/.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 diff --git a/.github/workflows/rust.yml b/.github/workflows/rust.yml index 9fd45e090..6dd5d7024 100644 --- a/.github/workflows/rust.yml +++ b/.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: Build - run: cargo build --verbose - - name: Run tests - run: cargo test --verbose + - name: cache protoc bin + id: cache-protoc-action + uses: actions/cache@v3 + env: + 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 diff --git a/Cargo.lock b/Cargo.lock index 22b19bba1..b8c76b3e2 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -447,6 +447,7 @@ dependencies = [ "http", "lazy_static", "netif", + "num_cpus", "openssl", "path-absolutize", "path-clean", diff --git a/Makefile b/Makefile new file mode 100644 index 000000000..6f05e92b5 --- /dev/null +++ b/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) diff --git a/ecstore/Cargo.toml b/ecstore/Cargo.toml index ada0fc7a5..75f503ef4 100644 --- a/ecstore/Cargo.toml +++ b/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" diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 7bf5cb5e6..9a85464dc 100644 --- a/ecstore/src/disk/local.rs +++ b/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( diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index 7853c5df4..95526ab73 100644 --- a/ecstore/src/sets.rs +++ b/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, @@ -254,8 +257,7 @@ impl StorageAPI for Sets { let mut set_obj_map = HashMap::new(); // hash key - let mut i = 0; - for obj in objects.iter() { + for (i, obj) in objects.iter().enumerate() { 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 { - if let Some(val) = set_obj_map.get_mut(&idx) { - val.push(DelObj { - // set_idx: idx, - orig_idx: i, - obj: obj.clone(), - }); - } + } else if let Some(val) = set_obj_map.get_mut(&idx) { + val.push(DelObj { + // set_idx: idx, + orig_idx: i, + 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 = v.iter().map(|v| v.obj.clone()).collect(); - let (dobjects, errs) = disks.delete_objects(bucket, objs, opts.clone()).await?; + let semaphore = semaphore.clone(); + let opts = opts.clone(); + let bucket = bucket.to_string(); - let mut i = 0; - for err in errs { - let obj = v.get(i).unwrap(); + let jh = tokio::spawn(async move { + let _permit = semaphore.acquire().await.unwrap(); + let objs: Vec = 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(); - - i += 1; - } + for (dobjects, errs) in results { + del_objects.extend(dobjects); + del_errs.extend(errs); } Ok((del_objects, del_errs)) diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index caa21174d..1454832d8 100644 --- a/ecstore/src/store.rs +++ b/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>>>; + lazy_static! { pub static ref GLOBAL_LOCAL_DISK_MAP: Arc>>> = Arc::new(RwLock::new(HashMap::new())); - pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc>>>>> = - Arc::new(RwLock::new(Vec::new())); + pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc> = Arc::new(RwLock::new(Vec::new())); } pub async fn find_local_disk(disk_path: &String) -> Option { @@ -75,7 +78,7 @@ pub async fn find_local_disk(disk_path: &String) -> Option { pub async fn all_local_disk_path() -> Vec { 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 { @@ -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() { - ress.push(fi.unwrap().to_object_info(&opts.bucket, &entry.name, false)); + if let Some(f) = fi { + 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)> { - let mut futures = Vec::new(); - - 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())) + internal_get_pool_info_existing_with_opts(&self.pools, bucket, object, opts).await } } +async fn internal_get_pool_info_existing_with_opts( + pools: &[Arc], + bucket: &str, + object: &str, + opts: &ObjectOptions, +) -> Result<(PoolObjInfo, Vec)> { + 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 opt = ObjectOptions::default(); - // 取所有poolObjInfo - 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 mut jhs = Vec::new(); + let semaphore = Arc::new(Semaphore::new(num_cpus::get())); + let pools = Arc::new(self.pools.clone()); - 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 res in results { + for (i, res) in results.into_iter().enumerate() { 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 = obj_idxs - .iter() - .filter_map(|&idx| { - if let Some(obj) = objects.get(idx) { - Some(obj.clone()) - } else { - None - } - }) - .collect(); + let objs: Vec = obj_idxs.iter().filter_map(|&idx| objects.get(idx).cloned()).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 err in perrs { + for (i, err) in perrs.into_iter().enumerate() { 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 { if opts.delete_prefix { - self.delete_prefix(bucket, &object).await?; + self.delete_prefix(bucket, object).await?; return Ok(ObjectInfo::default()); }