support distribute lock

Signed-off-by: junxiang Mu <1948535941@qq.com>

ecstore/src/bucket_meta.rs

@@ -2,7 +2,7 @@ use rmp_serde::Serializer;
 use serde::{Deserialize, Serialize};
 use time::OffsetDateTime;
 
-use crate::error::Result;
+use common::error::Result;
 
 use crate::disk::BUCKET_META_PREFIX;
 

ecstore/src/chunk_stream.rs

@@ -1,5 +1,5 @@
-use crate::error::StdError;
 use bytes::Bytes;
+use common::error::StdError;
 use futures::pin_mut;
 use futures::stream::{Stream, StreamExt};
 use std::future::Future;

ecstore/src/disk/endpoint.rs

@@ -1,5 +1,5 @@
-use crate::error::{Error, Result};
 use crate::utils::net;
+use common::error::{Error, Result};
 use path_absolutize::Absolutize;
 use path_clean::PathClean;
 use std::{fmt::Display, path::Path};

ecstore/src/disk/error.rs

@@ -1,4 +1,4 @@
-use crate::error::{Error, Result};
+use common::error::{Error, Result};
 
 #[derive(Debug, thiserror::Error)]
 pub enum DiskError {

ecstore/src/disk/format.rs

@@ -1,5 +1,5 @@
 use super::error::DiskError;
-use crate::error::{Error, Result};
+use common::error::{Error, Result};
 use serde::{Deserialize, Serialize};
 use serde_json::Error as JsonError;
 use uuid::Uuid;

ecstore/src/disk/local.rs

@@ -5,12 +5,12 @@ use super::{
 };
 use crate::disk::{LocalFileReader, LocalFileWriter, STORAGE_FORMAT_FILE};
 use crate::{
-    error::{Error, Result},
     file_meta::FileMeta,
     store_api::{FileInfo, RawFileInfo},
     utils,
 };
 use bytes::Bytes;
+use common::error::{Error, Result};
 use path_absolutize::Absolutize;
 use std::{
     fs::Metadata,

ecstore/src/disk/mod.rs

@@ -2,7 +2,7 @@ pub mod endpoint;
 pub mod error;
 pub mod format;
 mod local;
-mod remote;
+pub mod remote;
 
 pub const RUSTFS_META_BUCKET: &str = ".rustfs.sys";
 pub const RUSTFS_META_MULTIPART_BUCKET: &str = ".rustfs.sys/multipart";
@@ -14,11 +14,11 @@ const STORAGE_FORMAT_FILE: &str = "xl.meta";
 
 use crate::{
     erasure::{ReadAt, Write},
-    error::{Error, Result},
     file_meta::FileMeta,
     store_api::{FileInfo, RawFileInfo},
 };
 use bytes::Bytes;
+use common::error::{Error, Result};
 use futures::StreamExt;
 use protos::proto_gen::node_service::{
     node_service_client::NodeServiceClient, ReadAtRequest, ReadAtResponse, WriteRequest, WriteResponse,

ecstore/src/disk/remote.rs

@@ -1,6 +1,7 @@
 use std::{path::PathBuf, sync::Arc, time::Duration};
 
 use bytes::Bytes;
+use common::error::{Error, Result};
 use futures::lock::Mutex;
 use protos::{
     node_service_time_out_client,
@@ -12,7 +13,7 @@ use protos::{
     },
     DEFAULT_GRPC_SERVER_MESSAGE_LEN,
 };
-use tokio::{fs, sync::RwLock};
+use tokio::sync::RwLock;
 use tonic::{
     transport::{Channel, Endpoint as tonic_Endpoint},
     Request,
@@ -23,7 +24,6 @@ use uuid::Uuid;
 
 use crate::{
     disk::error::DiskError,
-    error::{Error, Result},
     store_api::{FileInfo, RawFileInfo},
 };
 
@@ -35,15 +35,16 @@ use super::{
 
 #[derive(Debug)]
 pub struct RemoteDisk {
-    id: Mutex<Option<Uuid>>,
-    channel: Arc<RwLock<Option<Channel>>>,
-    url: url::Url,
+    pub id: Mutex<Option<Uuid>>,
+    pub channel: Arc<RwLock<Option<Channel>>>,
+    pub url: url::Url,
     pub root: PathBuf,
 }
 
 impl RemoteDisk {
     pub async fn new(ep: &Endpoint, _opt: &DiskOption) -> Result<Self> {
-        let root = fs::canonicalize(ep.url.path()).await?;
+        // let root = fs::canonicalize(ep.url.path()).await?;
+        let root = PathBuf::from(ep.url.path());
 
         Ok(Self {
             channel: Arc::new(RwLock::new(None)),

ecstore/src/disks_layout.rs

@@ -1,5 +1,5 @@
-use crate::error::{Error, Result};
 use crate::utils::ellipses::*;
+use common::error::{Error, Result};
 use serde::Deserialize;
 use std::collections::HashSet;
 

ecstore/src/endpoints.rs

@@ -1,9 +1,9 @@
 use crate::{
     disk::endpoint::{Endpoint, EndpointType},
     disks_layout::DisksLayout,
-    error::{Error, Result},
     utils::net,
 };
+use common::error::{Error, Result};
 use std::{
     collections::{hash_map::Entry, HashMap, HashSet},
     net::IpAddr,

ecstore/src/erasure.rs

@@ -1,5 +1,5 @@
-use crate::error::{Error, Result, StdError};
 use bytes::Bytes;
+use common::error::{Error, Result, StdError};
 use futures::future::join_all;
 use futures::{Stream, StreamExt};
 use reed_solomon_erasure::galois_8::ReedSolomon;

ecstore/src/error.rs

@@ -1,82 +0,0 @@
-use tracing_error::{SpanTrace, SpanTraceStatus};
-
-pub type StdError = Box<dyn std::error::Error + Send + Sync + 'static>;
-
-pub type Result<T = (), E = Error> = std::result::Result<T, E>;
-
-#[derive(Debug)]
-pub struct Error {
-    inner: Box<dyn std::error::Error + Send + Sync + 'static>,
-    span_trace: SpanTrace,
-}
-
-impl Error {
-    /// Create a new error from a `std::error::Error`.
-    #[must_use]
-    #[track_caller]
-    pub fn new<T: std::error::Error + Send + Sync + 'static>(source: T) -> Self {
-        Self::from_std_error(source.into())
-    }
-
-    /// Create a new error from a `std::error::Error`.
-    #[must_use]
-    #[track_caller]
-    pub fn from_std_error(inner: StdError) -> Self {
-        Self {
-            inner,
-            span_trace: SpanTrace::capture(),
-        }
-    }
-
-    /// Create a new error from a string.
-    #[must_use]
-    #[track_caller]
-    pub fn from_string(s: impl Into<String>) -> Self {
-        Self::msg(s)
-    }
-
-    /// Create a new error from a string.
-    #[must_use]
-    #[track_caller]
-    pub fn msg(s: impl Into<String>) -> Self {
-        Self::from_std_error(s.into().into())
-    }
-
-    /// Returns `true` if the inner type is the same as `T`.
-    #[inline]
-    pub fn is<T: std::error::Error + 'static>(&self) -> bool {
-        self.inner.is::<T>()
-    }
-
-    /// Returns some reference to the inner value if it is of type `T`, or
-    /// `None` if it isn't.
-    #[inline]
-    pub fn downcast_ref<T: std::error::Error + 'static>(&self) -> Option<&T> {
-        self.inner.downcast_ref()
-    }
-
-    /// Returns some mutable reference to the inner value if it is of type `T`, or
-    /// `None` if it isn't.
-    #[inline]
-    pub fn downcast_mut<T: std::error::Error + 'static>(&mut self) -> Option<&mut T> {
-        self.inner.downcast_mut()
-    }
-}
-
-impl<T: std::error::Error + Send + Sync + 'static> From<T> for Error {
-    fn from(e: T) -> Self {
-        Self::new(e)
-    }
-}
-
-impl std::fmt::Display for Error {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        write!(f, "{}", self.inner)?;
-
-        if self.span_trace.status() != SpanTraceStatus::EMPTY {
-            write!(f, "\nspan_trace:\n{}", self.span_trace)?;
-        }
-
-        Ok(())
-    }
-}

ecstore/src/lib.rs

@@ -4,7 +4,6 @@ pub mod disk;
 pub mod disks_layout;
 pub mod endpoints;
 pub mod erasure;
-pub mod error;
 mod file_meta;
 pub mod peer;
 pub mod set_disk;

ecstore/src/peer.rs

@@ -1,4 +1,5 @@
 use async_trait::async_trait;
+use common::error::{Error, Result};
 use futures::future::join_all;
 use protos::proto_gen::node_service::node_service_client::NodeServiceClient;
 use protos::proto_gen::node_service::{DeleteBucketRequest, GetBucketInfoRequest, ListBucketRequest, MakeBucketRequest};
@@ -15,7 +16,6 @@ use crate::store::all_local_disk;
 use crate::{
     disk::{self, error::DiskError, VolumeInfo},
     endpoints::{EndpointServerPools, Node},
-    error::{Error, Result},
     store_api::{BucketInfo, BucketOptions, MakeBucketOptions},
 };
 

ecstore/src/sets.rs

@@ -1,7 +1,11 @@
-use std::collections::HashMap;
+use std::{collections::HashMap, sync::Arc};
 
+use common::error::{Error, Result};
+use common::globals::GLOBAL_Local_Node_Name;
 use futures::future::join_all;
 use http::HeaderMap;
+use lock::{namespace_lock::NsLockMap, new_lock_api, LockApi};
+use tokio::sync::RwLock;
 use uuid::Uuid;
 
 use crate::{
@@ -10,7 +14,6 @@ use crate::{
         DiskStore,
     },
     endpoints::PoolEndpoints,
-    error::{Error, Result},
     set_disk::SetDisks,
     store::{GLOBAL_IsDistErasure, GLOBAL_LOCAL_DISK_SET_DRIVES},
     store_api::{
@@ -26,6 +29,7 @@ pub struct Sets {
     // pub sets: Vec<Objects>,
     // pub disk_set: Vec<Vec<Option<DiskStore>>>, // [set_count_idx][set_drive_count_idx] = disk_idx
     pub disk_set: Vec<SetDisks>, // [set_count_idx][set_drive_count_idx] = disk_idx
+    pub lockers: Vec<Vec<LockApi>>,
     pub pool_idx: usize,
     pub endpoints: PoolEndpoints,
     pub format: FormatV3,
@@ -46,6 +50,20 @@ impl Sets {
         let set_count = fm.erasure.sets.len();
         let set_drive_count = fm.erasure.sets[0].len();
 
+        let mut unique: Vec<Vec<String>> = vec![vec![]; set_count];
+        let mut lockers: Vec<Vec<LockApi>> = vec![vec![]; set_count];
+        endpoints.endpoints.as_ref().iter().enumerate().for_each(|(idx, endpoint)| {
+            let set_idx = idx / set_drive_count;
+            if !unique[set_idx].contains(&endpoint.url.host_str().unwrap().to_string()) {
+                if endpoint.is_local {
+                    unique[set_idx].push(endpoint.url.host_str().unwrap().to_string());
+                    lockers[set_idx].push(new_lock_api(true, None));
+                } else {
+                    lockers[set_idx].push(new_lock_api(false, Some(endpoint.url.clone())));
+                }
+            }
+        });
+
         let mut disk_set = Vec::with_capacity(set_count);
 
         for i in 0..set_count {
@@ -83,6 +101,9 @@ impl Sets {
 
             let set_disks = SetDisks {
                 disks: set_drive,
+                lockers: lockers[i].clone(),
+                locker_owner: GLOBAL_Local_Node_Name.read().await.to_string(),
+                ns_mutex: Arc::new(RwLock::new(NsLockMap::default())),
                 set_drive_count,
                 parity_count: partiy_count,
                 set_index: i,
@@ -96,6 +117,7 @@ impl Sets {
             id: fm.id,
             // sets: todo!(),
             disk_set,
+            lockers,
             pool_idx,
             endpoints: endpoints.clone(),
             format: fm.clone(),

ecstore/src/store.rs

@@ -1,10 +1,10 @@
 use crate::{
     bucket_meta::BucketMetadata,
     disk::{
-        error::DiskError, new_disk, DeleteOptions, DiskOption, DiskStore, WalkDirOptions, BUCKET_META_PREFIX, RUSTFS_META_BUCKET,
+        endpoint::EndpointType, error::DiskError, new_disk, DeleteOptions, DiskOption, DiskStore, WalkDirOptions,
+        BUCKET_META_PREFIX, RUSTFS_META_BUCKET,
     },
     endpoints::{EndpointServerPools, SetupType},
-    error::{Error, Result},
     peer::S3PeerSys,
     sets::Sets,
     store_api::{
@@ -15,6 +15,8 @@ use crate::{
     store_init, utils,
 };
 use backon::{ExponentialBuilder, Retryable};
+use common::error::{Error, Result};
+use common::globals::{GLOBAL_Local_Node_Name, GLOBAL_Rustfs_Host, GLOBAL_Rustfs_Port};
 use futures::future::join_all;
 use http::HeaderMap;
 use s3s::{dto::StreamingBlob, Body};
@@ -170,6 +172,13 @@ impl ECStore {
 
         let mut local_disks = Vec::new();
 
+        init_local_peer(
+            &endpoint_pools,
+            &GLOBAL_Rustfs_Host.read().await.to_string(),
+            &GLOBAL_Rustfs_Port.read().await.to_string(),
+        )
+        .await;
+
         info!("endpoint_pools: {:?}", endpoint_pools);
 
         for (i, pool_eps) in endpoint_pools.as_ref().iter().enumerate() {
@@ -791,3 +800,24 @@ impl StorageAPI for ECStore {
         Ok(())
     }
 }
+
+async fn init_local_peer(endpoint_pools: &EndpointServerPools, host: &String, port: &String) {
+    let mut peer_set = Vec::new();
+    endpoint_pools.as_ref().iter().for_each(|endpoints| {
+        endpoints.endpoints.as_ref().iter().for_each(|endpoint| {
+            if endpoint.get_type() == EndpointType::Url && endpoint.is_local && endpoint.url.has_host() {
+                peer_set.push(endpoint.url.host_str().unwrap().to_string());
+            }
+        });
+    });
+
+    if peer_set.is_empty() {
+        if !host.is_empty() {
+            *GLOBAL_Local_Node_Name.write().await = format!("{}:{}", host, port);
+        }
+
+        *GLOBAL_Local_Node_Name.write().await = format!("127.0.0.1:{}", port);
+    }
+
+    *GLOBAL_Local_Node_Name.write().await = peer_set[0].clone();
+}

ecstore/src/store_api.rs

@@ -1,4 +1,4 @@
-use crate::error::{Error, Result};
+use common::error::{Error, Result};
 use http::HeaderMap;
 use rmp_serde::Serializer;
 use s3s::dto::StreamingBlob;
@@ -366,6 +366,7 @@ pub struct ObjectOptions {
 
     pub delete_prefix: bool,
     pub version_id: String,
+    pub no_lock: bool,
 }
 
 // impl Default for ObjectOptions {

ecstore/src/store_init.rs

@@ -5,8 +5,8 @@ use crate::{
         new_disk, DiskOption, DiskStore, FORMAT_CONFIG_FILE, RUSTFS_META_BUCKET,
     },
     endpoints::Endpoints,
-    error::{Error, Result},
 };
+use common::error::{Error, Result};
 use futures::future::join_all;
 use std::{
     collections::{hash_map::Entry, HashMap},

ecstore/src/utils/ellipses.rs

@@ -1,4 +1,4 @@
-use crate::error::{Error, Result};
+use common::error::{Error, Result};
 use lazy_static::*;
 use regex::Regex;
 

ecstore/src/utils/net.rs

@@ -1,4 +1,4 @@
-use crate::error::{Error, Result};
+use common::error::{Error, Result};
 use lazy_static::lazy_static;
 use std::{
     collections::HashSet,