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

refactor(lock): unify NamespaceLock client model and LockRequest API

Browse Source 
- Refactor NamespaceLock to use a unified client vector and quorum mechanism,
  removing legacy local/distributed lock split and related code.
- Update LockRequest to split timeout into acquire_timeout and ttl, and add
  builder methods for both.
- Adjust all batch lock APIs to accept ttl and use new LockRequest fields.
- Update all affected tests and documentation for the new API.

Signed-off-by: junxiang Mu <1948535941@qq.com>
This commit is contained in:
junxiang Mu
2025-07-23 16:00:46 +08:00
parent dc44cde081
commit cad005bc21
20 changed files with 1729 additions and 2668 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+44 -1
View File
@@ -3477,6 +3477,7 @@ dependencies = [
"rustfs-protos",
"serde",
"serde_json",
"serial_test",
"tokio",
"tonic",
"url",
@@ -8139,7 +8140,9 @@ name = "rustfs-lock"
version = "0.0.5"
dependencies = [
"async-trait",
"dashmap 6.1.0",
"bytes",
"futures",
"lazy_static",
"lru 0.16.0",
"once_cell",
"rand 0.9.2",
@@ -8663,6 +8666,15 @@ dependencies = [
"winapi-util",
]
[[package]]
name = "scc"
version = "2.3.4"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "22b2d775fb28f245817589471dd49c5edf64237f4a19d10ce9a92ff4651a27f4"
dependencies = [
"sdd",
]
[[package]]
name = "schannel"
version = "0.1.27"
@@ -8694,6 +8706,12 @@ dependencies = [
"untrusted",
]
[[package]]
name = "sdd"
version = "3.0.10"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "490dcfcbfef26be6800d11870ff2df8774fa6e86d047e3e8c8a76b25655e41ca"
[[package]]
name = "sec1"
version = "0.3.0"
@@ -8951,6 +8969,31 @@ dependencies = [
"unsafe-libyaml",
]
[[package]]
name = "serial_test"
version = "3.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "1b258109f244e1d6891bf1053a55d63a5cd4f8f4c30cf9a1280989f80e7a1fa9"
dependencies = [
"futures",
"log",
"once_cell",
"parking_lot",
"scc",
"serial_test_derive",
]
[[package]]
name = "serial_test_derive"
version = "3.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "5d69265a08751de7844521fd15003ae0a888e035773ba05695c5c759a6f89eef"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.104",
]
[[package]]
name = "server_fn"
version = "0.6.15"
crates/e2e_test/Cargo.toml
+1
View File
@@ -38,3 +38,4 @@ url.workspace = true
rustfs-madmin.workspace = true
rustfs-filemeta.workspace = true
bytes.workspace = true
serial_test = "3.2.0"
crates/e2e_test/src/reliant/lock.rs
+178 -179
View File
@@ -13,23 +13,42 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use rustfs_lock::{create_namespace_lock, LockArgs, NamespaceLockManager};
use rustfs_ecstore::{disk::endpoint::Endpoint, lock_utils::create_unique_clients};
use rustfs_lock::{LockId, LockMetadata, LockPriority, LockType};
use rustfs_lock::{LockRequest, NamespaceLock, NamespaceLockManager};
use rustfs_protos::{node_service_time_out_client, proto_gen::node_service::GenerallyLockRequest};
use serial_test::serial;
use std::{error::Error, time::Duration};
use tokio::time::sleep;
use tonic::Request;
use url::Url;
const CLUSTER_ADDR: &str = "http://localhost:9000";
fn get_cluster_endpoints() -> Vec<Endpoint> {
vec![Endpoint {
url: Url::parse(CLUSTER_ADDR).unwrap(),
is_local: false,
pool_idx: 0,
set_idx: 0,
disk_idx: 0,
}]
}
#[tokio::test]
#[serial]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_lock_unlock_rpc() -> Result<(), Box<dyn Error>> {
let args = LockArgs {
uid: "1111".to_string(),
resources: vec!["dandan".to_string()],
let args = LockRequest {
lock_id: LockId::new_deterministic("dandan"),
resource: "dandan".to_string(),
lock_type: LockType::Exclusive,
owner: "dd".to_string(),
source: "".to_string(),
quorum: 3,
acquire_timeout: Duration::from_secs(30),
ttl: Duration::from_secs(30),
metadata: LockMetadata::default(),
priority: LockPriority::Normal,
deadlock_detection: false,
};
let args = serde_json::to_string(&args)?;
@@ -54,39 +73,51 @@ async fn test_lock_unlock_rpc() -> Result<(), Box<dyn Error>> {
}
#[tokio::test]
#[serial]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_lock_unlock_ns_lock() -> Result<(), Box<dyn Error>> {
let ns_lock = create_namespace_lock("test".to_string(), true);
let endpoints = get_cluster_endpoints();
let clients = create_unique_clients(&endpoints).await?;
let ns_lock = NamespaceLock::with_clients("test".to_string(), clients);
let resources = vec!["foo".to_string()];
let result = ns_lock.lock_batch(&resources, "dandan", Duration::from_secs(5)).await;
let result = ns_lock
.lock_batch(&resources, "dandan", Duration::from_secs(5), Duration::from_secs(10))
.await;
match &result {
Ok(success) => println!("Lock result: {}", success),
Err(e) => println!("Lock error: {}", e),
Ok(success) => println!("Lock result: {success}"),
Err(e) => println!("Lock error: {e}"),
}
let result = result?;
assert!(result, "Lock should succeed, but got: {}", result);
assert!(result, "Lock should succeed, but got: {result}");
ns_lock.unlock_batch(&resources, "dandan").await?;
Ok(())
}
#[tokio::test]
#[serial]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_concurrent_lock_attempts() -> Result<(), Box<dyn Error>> {
let ns_lock = create_namespace_lock("test".to_string(), true);
let endpoints = get_cluster_endpoints();
let clients = create_unique_clients(&endpoints).await?;
let ns_lock = NamespaceLock::with_clients("test".to_string(), clients);
let resource = vec!["concurrent_resource".to_string()];
// First lock should succeed
println!("Attempting first lock...");
let result1 = ns_lock.lock_batch(&resource, "owner1", Duration::from_secs(5)).await?;
println!("First lock result: {}", result1);
let result1 = ns_lock
.lock_batch(&resource, "owner1", Duration::from_secs(5), Duration::from_secs(10))
.await?;
println!("First lock result: {result1}");
assert!(result1, "First lock should succeed");
// Second lock should fail (resource already locked)
println!("Attempting second lock...");
let result2 = ns_lock.lock_batch(&resource, "owner2", Duration::from_secs(1)).await?;
println!("Second lock result: {}", result2);
let result2 = ns_lock
.lock_batch(&resource, "owner2", Duration::from_secs(1), Duration::from_secs(10))
.await?;
println!("Second lock result: {result2}");
assert!(!result2, "Second lock should fail");
// Unlock by first owner
@@ -96,8 +127,10 @@ async fn test_concurrent_lock_attempts() -> Result<(), Box<dyn Error>> {
// Now second owner should be able to lock
println!("Attempting third lock...");
let result3 = ns_lock.lock_batch(&resource, "owner2", Duration::from_secs(5)).await?;
println!("Third lock result: {}", result3);
let result3 = ns_lock
.lock_batch(&resource, "owner2", Duration::from_secs(5), Duration::from_secs(10))
.await?;
println!("Third lock result: {result3}");
assert!(result3, "Lock should succeed after unlock");
// Clean up
@@ -109,21 +142,30 @@ async fn test_concurrent_lock_attempts() -> Result<(), Box<dyn Error>> {
}
#[tokio::test]
#[serial]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_read_write_lock_compatibility() -> Result<(), Box<dyn Error>> {
let ns_lock = create_namespace_lock("test_rw".to_string(), true);
let endpoints = get_cluster_endpoints();
let clients = create_unique_clients(&endpoints).await?;
let ns_lock = NamespaceLock::with_clients("test_rw".to_string(), clients);
let resource = vec!["rw_resource".to_string()];
// First read lock should succeed
let result1 = ns_lock.rlock_batch(&resource, "reader1", Duration::from_secs(5)).await?;
let result1 = ns_lock
.rlock_batch(&resource, "reader1", Duration::from_secs(5), Duration::from_secs(10))
.await?;
assert!(result1, "First read lock should succeed");
// Second read lock should also succeed (read locks are compatible)
let result2 = ns_lock.rlock_batch(&resource, "reader2", Duration::from_secs(5)).await?;
let result2 = ns_lock
.rlock_batch(&resource, "reader2", Duration::from_secs(5), Duration::from_secs(10))
.await?;
assert!(result2, "Second read lock should succeed");
// Write lock should fail (read locks are held)
let result3 = ns_lock.lock_batch(&resource, "writer1", Duration::from_secs(1)).await?;
let result3 = ns_lock
.lock_batch(&resource, "writer1", Duration::from_secs(1), Duration::from_secs(10))
.await?;
assert!(!result3, "Write lock should fail when read locks are held");
// Release read locks
@@ -131,7 +173,9 @@ async fn test_read_write_lock_compatibility() -> Result<(), Box<dyn Error>> {
ns_lock.runlock_batch(&resource, "reader2").await?;
// Now write lock should succeed
let result4 = ns_lock.lock_batch(&resource, "writer1", Duration::from_secs(5)).await?;
let result4 = ns_lock
.lock_batch(&resource, "writer1", Duration::from_secs(5), Duration::from_secs(10))
.await?;
assert!(result4, "Write lock should succeed after read locks released");
// Clean up
@@ -141,20 +185,27 @@ async fn test_read_write_lock_compatibility() -> Result<(), Box<dyn Error>> {
}
#[tokio::test]
#[serial]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_lock_timeout() -> Result<(), Box<dyn Error>> {
let ns_lock = create_namespace_lock("test_timeout".to_string(), true);
let endpoints = get_cluster_endpoints();
let clients = create_unique_clients(&endpoints).await?;
let ns_lock = NamespaceLock::with_clients("test_timeout".to_string(), clients);
let resource = vec!["timeout_resource".to_string()];
// First lock with short timeout
let result1 = ns_lock.lock_batch(&resource, "owner1", Duration::from_secs(2)).await?;
let result1 = ns_lock
.lock_batch(&resource, "owner1", Duration::from_secs(2), Duration::from_secs(1))
.await?;
assert!(result1, "First lock should succeed");
// Wait for lock to expire
sleep(Duration::from_secs(3)).await;
sleep(Duration::from_secs(5)).await;
// Second lock should succeed after timeout
let result2 = ns_lock.lock_batch(&resource, "owner2", Duration::from_secs(5)).await?;
let result2 = ns_lock
.lock_batch(&resource, "owner2", Duration::from_secs(5), Duration::from_secs(1))
.await?;
assert!(result2, "Lock should succeed after timeout");
// Clean up
@@ -164,9 +215,12 @@ async fn test_lock_timeout() -> Result<(), Box<dyn Error>> {
}
#[tokio::test]
#[serial]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_batch_lock_operations() -> Result<(), Box<dyn Error>> {
let ns_lock = create_namespace_lock("test_batch".to_string(), true);
let endpoints = get_cluster_endpoints();
let clients = create_unique_clients(&endpoints).await?;
let ns_lock = NamespaceLock::with_clients("test_batch".to_string(), clients);
let resources = vec![
"batch_resource1".to_string(),
"batch_resource2".to_string(),
@@ -174,19 +228,25 @@ async fn test_batch_lock_operations() -> Result<(), Box<dyn Error>> {
];
// Lock all resources
let result = ns_lock.lock_batch(&resources, "batch_owner", Duration::from_secs(5)).await?;
let result = ns_lock
.lock_batch(&resources, "batch_owner", Duration::from_secs(5), Duration::from_secs(10))
.await?;
assert!(result, "Batch lock should succeed");
// Try to lock one of the resources with different owner - should fail
let single_resource = vec!["batch_resource2".to_string()];
let result2 = ns_lock.lock_batch(&single_resource, "other_owner", Duration::from_secs(1)).await?;
let result2 = ns_lock
.lock_batch(&single_resource, "other_owner", Duration::from_secs(1), Duration::from_secs(10))
.await?;
assert!(!result2, "Lock should fail for already locked resource");
// Unlock all resources
ns_lock.unlock_batch(&resources, "batch_owner").await?;
// Now should be able to lock single resource
let result3 = ns_lock.lock_batch(&single_resource, "other_owner", Duration::from_secs(5)).await?;
let result3 = ns_lock
.lock_batch(&single_resource, "other_owner", Duration::from_secs(5), Duration::from_secs(10))
.await?;
assert!(result3, "Lock should succeed after batch unlock");
// Clean up
@@ -196,17 +256,24 @@ async fn test_batch_lock_operations() -> Result<(), Box<dyn Error>> {
}
#[tokio::test]
#[serial]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_multiple_namespaces() -> Result<(), Box<dyn Error>> {
let ns_lock1 = create_namespace_lock("namespace1".to_string(), true);
let ns_lock2 = create_namespace_lock("namespace2".to_string(), true);
let endpoints = get_cluster_endpoints();
let clients = create_unique_clients(&endpoints).await?;
let ns_lock1 = NamespaceLock::with_clients("namespace1".to_string(), clients.clone());
let ns_lock2 = NamespaceLock::with_clients("namespace2".to_string(), clients);
let resource = vec!["shared_resource".to_string()];
// Lock same resource in different namespaces - both should succeed
let result1 = ns_lock1.lock_batch(&resource, "owner1", Duration::from_secs(5)).await?;
let result1 = ns_lock1
.lock_batch(&resource, "owner1", Duration::from_secs(5), Duration::from_secs(10))
.await?;
assert!(result1, "Lock in namespace1 should succeed");
let result2 = ns_lock2.lock_batch(&resource, "owner2", Duration::from_secs(5)).await?;
let result2 = ns_lock2
.lock_batch(&resource, "owner2", Duration::from_secs(5), Duration::from_secs(10))
.await?;
assert!(result2, "Lock in namespace2 should succeed");
// Clean up
@@ -217,19 +284,24 @@ async fn test_multiple_namespaces() -> Result<(), Box<dyn Error>> {
}
#[tokio::test]
#[serial]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_rpc_read_lock() -> Result<(), Box<dyn Error>> {
let args = LockArgs {
uid: "2222".to_string(),
resources: vec!["read_resource".to_string()],
let args = LockRequest {
lock_id: LockId::new_deterministic("read_resource"),
resource: "read_resource".to_string(),
lock_type: LockType::Shared,
owner: "reader1".to_string(),
source: "".to_string(),
quorum: 3,
acquire_timeout: Duration::from_secs(30),
ttl: Duration::from_secs(30),
metadata: LockMetadata::default(),
priority: LockPriority::Normal,
deadlock_detection: false,
};
let args_str = serde_json::to_string(&args)?;
let mut client = node_service_time_out_client(&CLUSTER_ADDR.to_string()).await?;
// First read lock
let request = Request::new(GenerallyLockRequest { args: args_str.clone() });
let response = client.r_lock(request).await?.into_inner();
@@ -238,12 +310,16 @@ async fn test_rpc_read_lock() -> Result<(), Box<dyn Error>> {
}
// Second read lock with different owner should also succeed
let args2 = LockArgs {
uid: "3333".to_string(),
resources: vec!["read_resource".to_string()],
let args2 = LockRequest {
lock_id: LockId::new_deterministic("read_resource"),
resource: "read_resource".to_string(),
lock_type: LockType::Shared,
owner: "reader2".to_string(),
source: "".to_string(),
quorum: 3,
acquire_timeout: Duration::from_secs(30),
ttl: Duration::from_secs(30),
metadata: LockMetadata::default(),
priority: LockPriority::Normal,
deadlock_detection: false,
};
let args2_str = serde_json::to_string(&args2)?;
let request2 = Request::new(GenerallyLockRequest { args: args2_str });
@@ -263,19 +339,24 @@ async fn test_rpc_read_lock() -> Result<(), Box<dyn Error>> {
}
#[tokio::test]
#[serial]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_lock_refresh() -> Result<(), Box<dyn Error>> {
let args = LockArgs {
uid: "4444".to_string(),
resources: vec!["refresh_resource".to_string()],
let args = LockRequest {
lock_id: LockId::new_deterministic("refresh_resource"),
resource: "refresh_resource".to_string(),
lock_type: LockType::Exclusive,
owner: "refresh_owner".to_string(),
source: "".to_string(),
quorum: 3,
acquire_timeout: Duration::from_secs(30),
ttl: Duration::from_secs(30),
metadata: LockMetadata::default(),
priority: LockPriority::Normal,
deadlock_detection: false,
};
let args_str = serde_json::to_string(&args)?;
let mut client = node_service_time_out_client(&CLUSTER_ADDR.to_string()).await?;
// Acquire lock
let request = Request::new(GenerallyLockRequest { args: args_str.clone() });
let response = client.lock(request).await?.into_inner();
@@ -302,19 +383,24 @@ async fn test_lock_refresh() -> Result<(), Box<dyn Error>> {
}
#[tokio::test]
#[serial]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_force_unlock() -> Result<(), Box<dyn Error>> {
let args = LockArgs {
uid: "5555".to_string(),
resources: vec!["force_resource".to_string()],
let args = LockRequest {
lock_id: LockId::new_deterministic("force_resource"),
resource: "force_resource".to_string(),
lock_type: LockType::Exclusive,
owner: "force_owner".to_string(),
source: "".to_string(),
quorum: 3,
acquire_timeout: Duration::from_secs(30),
ttl: Duration::from_secs(30),
metadata: LockMetadata::default(),
priority: LockPriority::Normal,
deadlock_detection: false,
};
let args_str = serde_json::to_string(&args)?;
let mut client = node_service_time_out_client(&CLUSTER_ADDR.to_string()).await?;
// Acquire lock
let request = Request::new(GenerallyLockRequest { args: args_str.clone() });
let response = client.lock(request).await?.into_inner();
@@ -323,12 +409,16 @@ async fn test_force_unlock() -> Result<(), Box<dyn Error>> {
}
// Force unlock (even by different owner)
let force_args = LockArgs {
uid: "5555".to_string(),
resources: vec!["force_resource".to_string()],
let force_args = LockRequest {
lock_id: LockId::new_deterministic("force_resource"),
resource: "force_resource".to_string(),
lock_type: LockType::Exclusive,
owner: "admin".to_string(),
source: "".to_string(),
quorum: 3,
acquire_timeout: Duration::from_secs(30),
ttl: Duration::from_secs(30),
metadata: LockMetadata::default(),
priority: LockPriority::Normal,
deadlock_detection: false,
};
let force_args_str = serde_json::to_string(&force_args)?;
let request = Request::new(GenerallyLockRequest { args: force_args_str });
@@ -342,137 +432,46 @@ async fn test_force_unlock() -> Result<(), Box<dyn Error>> {
}
#[tokio::test]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_concurrent_rpc_lock_attempts() -> Result<(), Box<dyn Error>> {
let args1 = LockArgs {
uid: "concurrent_test_1".to_string(),
resources: vec!["concurrent_rpc_resource".to_string()],
owner: "owner1".to_string(),
source: "".to_string(),
quorum: 3,
};
let args1_str = serde_json::to_string(&args1)?;
let args2 = LockArgs {
uid: "concurrent_test_2".to_string(),
resources: vec!["concurrent_rpc_resource".to_string()],
owner: "owner2".to_string(),
source: "".to_string(),
quorum: 3,
};
let args2_str = serde_json::to_string(&args2)?;
let mut client = node_service_time_out_client(&CLUSTER_ADDR.to_string()).await?;
// First lock should succeed
println!("Attempting first RPC lock...");
let request1 = Request::new(GenerallyLockRequest { args: args1_str.clone() });
let response1 = client.lock(request1).await?.into_inner();
println!("First RPC lock response: success={}, error={:?}", response1.success, response1.error_info);
assert!(response1.success && response1.error_info.is_none(), "First lock should succeed");
// Second lock should fail (resource already locked)
println!("Attempting second RPC lock...");
let request2 = Request::new(GenerallyLockRequest { args: args2_str });
let response2 = client.lock(request2).await?.into_inner();
println!("Second RPC lock response: success={}, error={:?}", response2.success, response2.error_info);
assert!(!response2.success, "Second lock should fail");
// Unlock by first owner
println!("Unlocking first RPC lock...");
let unlock_request = Request::new(GenerallyLockRequest { args: args1_str });
let unlock_response = client.un_lock(unlock_request).await?.into_inner();
println!("Unlock response: success={}, error={:?}", unlock_response.success, unlock_response.error_info);
assert!(unlock_response.success && unlock_response.error_info.is_none(), "Unlock should succeed");
Ok(())
}
#[tokio::test]
#[serial]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_global_lock_map_sharing() -> Result<(), Box<dyn Error>> {
// Create two separate NsLockMap instances
let lock_map1 = rustfs_lock::NsLockMap::new(false, None);
let lock_map2 = rustfs_lock::NsLockMap::new(false, None);
let endpoints = get_cluster_endpoints();
let clients = create_unique_clients(&endpoints).await?;
let ns_lock1 = NamespaceLock::with_clients("global_test".to_string(), clients.clone());
let ns_lock2 = NamespaceLock::with_clients("global_test".to_string(), clients);
let resource = vec!["global_test_resource".to_string()];
// First instance acquires lock
println!("First lock map attempting to acquire lock...");
let result1 = lock_map1.lock_batch_with_ttl(&resource, "owner1", std::time::Duration::from_secs(5), Some(std::time::Duration::from_secs(30))).await?;
println!("First lock result: {}", result1);
let result1 = ns_lock1
.lock_batch(&resource, "owner1", std::time::Duration::from_secs(5), std::time::Duration::from_secs(10))
.await?;
println!("First lock result: {result1}");
assert!(result1, "First lock should succeed");
// Second instance should fail to acquire the same lock
println!("Second lock map attempting to acquire lock...");
let result2 = lock_map2.lock_batch_with_ttl(&resource, "owner2", std::time::Duration::from_secs(1), Some(std::time::Duration::from_secs(30))).await?;
println!("Second lock result: {}", result2);
let result2 = ns_lock2
.lock_batch(&resource, "owner2", std::time::Duration::from_secs(1), std::time::Duration::from_secs(10))
.await?;
println!("Second lock result: {result2}");
assert!(!result2, "Second lock should fail because resource is already locked");
// Release lock from first instance
println!("First lock map releasing lock...");
lock_map1.unlock_batch(&resource, "owner1").await?;
ns_lock1.unlock_batch(&resource, "owner1").await?;
// Now second instance should be able to acquire lock
println!("Second lock map attempting to acquire lock again...");
let result3 = lock_map2.lock_batch_with_ttl(&resource, "owner2", std::time::Duration::from_secs(5), Some(std::time::Duration::from_secs(30))).await?;
println!("Third lock result: {}", result3);
let result3 = ns_lock2
.lock_batch(&resource, "owner2", std::time::Duration::from_secs(5), std::time::Duration::from_secs(10))
.await?;
println!("Third lock result: {result3}");
assert!(result3, "Lock should succeed after first lock is released");
// Clean up
lock_map2.unlock_batch(&resource, "owner2").await?;
Ok(())
}
#[tokio::test]
#[ignore = "requires running RustFS server at localhost:9000"]
async fn test_sequential_rpc_lock_calls() -> Result<(), Box<dyn Error>> {
let args = LockArgs {
uid: "sequential_test".to_string(),
resources: vec!["sequential_resource".to_string()],
owner: "sequential_owner".to_string(),
source: "".to_string(),
quorum: 3,
};
let args_str = serde_json::to_string(&args)?;
let mut client = node_service_time_out_client(&CLUSTER_ADDR.to_string()).await?;
// First lock should succeed
println!("First lock attempt...");
let request1 = Request::new(GenerallyLockRequest { args: args_str.clone() });
let response1 = client.lock(request1).await?.into_inner();
println!("First response: success={}, error={:?}", response1.success, response1.error_info);
assert!(response1.success && response1.error_info.is_none(), "First lock should succeed");
// Second lock with same owner should also succeed (re-entrant)
println!("Second lock attempt with same owner...");
let request2 = Request::new(GenerallyLockRequest { args: args_str.clone() });
let response2 = client.lock(request2).await?.into_inner();
println!("Second response: success={}, error={:?}", response2.success, response2.error_info);
// Different owner should fail
let args2 = LockArgs {
uid: "sequential_test_2".to_string(),
resources: vec!["sequential_resource".to_string()],
owner: "different_owner".to_string(),
source: "".to_string(),
quorum: 3,
};
let args2_str = serde_json::to_string(&args2)?;
println!("Third lock attempt with different owner...");
let request3 = Request::new(GenerallyLockRequest { args: args2_str });
let response3 = client.lock(request3).await?.into_inner();
println!("Third response: success={}, error={:?}", response3.success, response3.error_info);
assert!(!response3.success, "Lock with different owner should fail");
// Unlock
println!("Unlocking...");
let unlock_request = Request::new(GenerallyLockRequest { args: args_str });
let unlock_response = client.un_lock(unlock_request).await?.into_inner();
println!("Unlock response: success={}, error={:?}", unlock_response.success, unlock_response.error_info);
ns_lock2.unlock_batch(&resource, "owner2").await?;
Ok(())
}
crates/ecstore/src/error.rs
+6
View File
@@ -183,6 +183,9 @@ pub enum StorageError {
#[error("Io error: {0}")]
Io(std::io::Error),
#[error("Lock error: {0}")]
Lock(#[from] rustfs_lock::LockError),
}
impl StorageError {
@@ -409,6 +412,7 @@ impl Clone for StorageError {
StorageError::FirstDiskWait => StorageError::FirstDiskWait,
StorageError::TooManyOpenFiles => StorageError::TooManyOpenFiles,
StorageError::NoHealRequired => StorageError::NoHealRequired,
StorageError::Lock(e) => StorageError::Lock(e.clone()),
}
}
}
@@ -471,6 +475,7 @@ impl StorageError {
StorageError::ConfigNotFound => 0x35,
StorageError::TooManyOpenFiles => 0x36,
StorageError::NoHealRequired => 0x37,
StorageError::Lock(_) => 0x38,
}
}
@@ -535,6 +540,7 @@ impl StorageError {
0x35 => Some(StorageError::ConfigNotFound),
0x36 => Some(StorageError::TooManyOpenFiles),
0x37 => Some(StorageError::NoHealRequired),
0x38 => Some(StorageError::Lock(rustfs_lock::LockError::internal("Generic lock error".to_string()))),
_ => None,
}
}
crates/ecstore/src/lib.rs
+1
View File
@@ -30,6 +30,7 @@ pub mod erasure_coding;
pub mod error;
pub mod global;
pub mod heal;
pub mod lock_utils;
pub mod metrics_realtime;
pub mod notification_sys;
pub mod pools;
crates/ecstore/src/rpc/mod.rs
+1 -1
View File
@@ -22,4 +22,4 @@ pub use http_auth::{build_auth_headers, verify_rpc_signature};
pub use peer_rest_client::PeerRestClient;
pub use peer_s3_client::{LocalPeerS3Client, PeerS3Client, RemotePeerS3Client, S3PeerSys};
pub use remote_disk::RemoteDisk;
pub use tonic_service::{make_server, NodeService};
pub use tonic_service::{NodeService, make_server};
crates/ecstore/src/rpc/tonic_service.rs
+37 -79
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use std::{collections::HashMap, io::Cursor, pin::Pin};
use std::{collections::HashMap, io::Cursor, pin::Pin, sync::Arc};
// use common::error::Error as EcsError;
use crate::{
@@ -36,6 +36,7 @@ use futures::{Stream, StreamExt};
use futures_util::future::join_all;
use rustfs_common::globals::GLOBAL_Local_Node_Name;
use rustfs_lock::{LockClient, LockRequest};
use bytes::Bytes;
use rmp_serde::{Deserializer, Serializer};
@@ -80,12 +81,12 @@ type ResponseStream<T> = Pin<Box<dyn Stream<Item = Result<T, tonic::Status>> + S
#[derive(Debug)]
pub struct NodeService {
local_peer: LocalPeerS3Client,
lock_manager: rustfs_lock::NsLockMap,
lock_manager: Arc<rustfs_lock::LocalClient>,
}
pub fn make_server() -> NodeService {
let local_peer = LocalPeerS3Client::new(None, None);
let lock_manager = rustfs_lock::NsLockMap::new(false, None);
let lock_manager = Arc::new(rustfs_lock::LocalClient::new());
NodeService {
local_peer,
lock_manager,
@@ -1531,7 +1532,7 @@ impl Node for NodeService {
async fn lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
let request = request.into_inner();
// Parse the request to extract resource and owner
let args: serde_json::Value = match serde_json::from_str(&request.args) {
let args: LockRequest = match serde_json::from_str(&request.args) {
Ok(args) => args,
Err(err) => {
return Ok(tonic::Response::new(GenerallyLockResponse {
@@ -1541,35 +1542,24 @@ impl Node for NodeService {
}
};
let resource = args["resources"][0].as_str().unwrap_or("");
let owner = args["owner"].as_str().unwrap_or("");
if resource.is_empty() {
return Ok(tonic::Response::new(GenerallyLockResponse {
success: false,
error_info: Some("No resource specified".to_string()),
}));
}
match self
.lock_manager
.lock_batch_with_ttl(&[resource.to_string()], owner, std::time::Duration::from_secs(30), Some(std::time::Duration::from_secs(30)))
.await
{
match self.lock_manager.acquire_exclusive(&args).await {
Ok(result) => Ok(tonic::Response::new(GenerallyLockResponse {
success: result,
success: result.success,
error_info: None,
})),
Err(err) => Ok(tonic::Response::new(GenerallyLockResponse {
success: false,
error_info: Some(format!("can not lock, resource: {resource}, owner: {owner}, err: {err}")),
error_info: Some(format!(
"can not lock, resource: {0}, owner: {1}, err: {2}",
args.resource, args.owner, err
)),
})),
}
}
async fn un_lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
let request = request.into_inner();
let args: serde_json::Value = match serde_json::from_str(&request.args) {
let args: LockRequest = match serde_json::from_str(&request.args) {
Ok(args) => args,
Err(err) => {
return Ok(tonic::Response::new(GenerallyLockResponse {
@@ -1579,31 +1569,24 @@ impl Node for NodeService {
}
};
let resource = args["resources"][0].as_str().unwrap_or("");
let owner = args["owner"].as_str().unwrap_or("");
if resource.is_empty() {
return Ok(tonic::Response::new(GenerallyLockResponse {
success: false,
error_info: Some("No resource specified".to_string()),
}));
}
match self.lock_manager.unlock_batch(&[resource.to_string()], owner).await {
match self.lock_manager.release(&args.lock_id).await {
Ok(_) => Ok(tonic::Response::new(GenerallyLockResponse {
success: true,
error_info: None,
})),
Err(err) => Ok(tonic::Response::new(GenerallyLockResponse {
success: false,
error_info: Some(format!("can not unlock, resource: {resource}, owner: {owner}, err: {err}")),
error_info: Some(format!(
"can not unlock, resource: {0}, owner: {1}, err: {2}",
args.resource, args.owner, err
)),
})),
}
}
async fn r_lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
let request = request.into_inner();
let args: serde_json::Value = match serde_json::from_str(&request.args) {
let args: LockRequest = match serde_json::from_str(&request.args) {
Ok(args) => args,
Err(err) => {
return Ok(tonic::Response::new(GenerallyLockResponse {
@@ -1613,35 +1596,24 @@ impl Node for NodeService {
}
};
let resource = args["resources"][0].as_str().unwrap_or("");
let owner = args["owner"].as_str().unwrap_or("");
if resource.is_empty() {
return Ok(tonic::Response::new(GenerallyLockResponse {
success: false,
error_info: Some("No resource specified".to_string()),
}));
}
match self
.lock_manager
.rlock_batch_with_ttl(&[resource.to_string()], owner, std::time::Duration::from_secs(30), Some(std::time::Duration::from_secs(30)))
.await
{
match self.lock_manager.acquire_shared(&args).await {
Ok(result) => Ok(tonic::Response::new(GenerallyLockResponse {
success: result,
success: result.success,
error_info: None,
})),
Err(err) => Ok(tonic::Response::new(GenerallyLockResponse {
success: false,
error_info: Some(format!("can not rlock, resource: {resource}, owner: {owner}, err: {err}")),
error_info: Some(format!(
"can not rlock, resource: {0}, owner: {1}, err: {2}",
args.resource, args.owner, err
)),
})),
}
}
async fn r_un_lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
let request = request.into_inner();
let args: serde_json::Value = match serde_json::from_str(&request.args) {
let args: LockRequest = match serde_json::from_str(&request.args) {
Ok(args) => args,
Err(err) => {
return Ok(tonic::Response::new(GenerallyLockResponse {
@@ -1651,31 +1623,24 @@ impl Node for NodeService {
}
};
let resource = args["resources"][0].as_str().unwrap_or("");
let owner = args["owner"].as_str().unwrap_or("");
if resource.is_empty() {
return Ok(tonic::Response::new(GenerallyLockResponse {
success: false,
error_info: Some("No resource specified".to_string()),
}));
}
match self.lock_manager.runlock_batch(&[resource.to_string()], owner).await {
match self.lock_manager.release(&args.lock_id).await {
Ok(_) => Ok(tonic::Response::new(GenerallyLockResponse {
success: true,
error_info: None,
})),
Err(err) => Ok(tonic::Response::new(GenerallyLockResponse {
success: false,
error_info: Some(format!("can not runlock, resource: {resource}, owner: {owner}, err: {err}")),
error_info: Some(format!(
"can not runlock, resource: {0}, owner: {1}, err: {2}",
args.resource, args.owner, err
)),
})),
}
}
async fn force_un_lock(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
let request = request.into_inner();
let args: serde_json::Value = match serde_json::from_str(&request.args) {
let args: LockRequest = match serde_json::from_str(&request.args) {
Ok(args) => args,
Err(err) => {
return Ok(tonic::Response::new(GenerallyLockResponse {
@@ -1685,31 +1650,24 @@ impl Node for NodeService {
}
};
let resource = args["resources"][0].as_str().unwrap_or("");
let owner = args["owner"].as_str().unwrap_or("");
if resource.is_empty() {
return Ok(tonic::Response::new(GenerallyLockResponse {
success: false,
error_info: Some("No resource specified".to_string()),
}));
}
match self.lock_manager.unlock_batch(&[resource.to_string()], owner).await {
match self.lock_manager.release(&args.lock_id).await {
Ok(_) => Ok(tonic::Response::new(GenerallyLockResponse {
success: true,
error_info: None,
})),
Err(err) => Ok(tonic::Response::new(GenerallyLockResponse {
success: false,
error_info: Some(format!("can not force_unlock, resource: {resource}, owner: {owner}, err: {err}")),
error_info: Some(format!(
"can not force_unlock, resource: {0}, owner: {1}, err: {2}",
args.resource, args.owner, err
)),
})),
}
}
async fn refresh(&self, request: Request<GenerallyLockRequest>) -> Result<Response<GenerallyLockResponse>, Status> {
let request = request.into_inner();
let _args: serde_json::Value = match serde_json::from_str(&request.args) {
let _args: LockRequest = match serde_json::from_str(&request.args) {
Ok(args) => args,
Err(err) => {
return Ok(tonic::Response::new(GenerallyLockResponse {
crates/ecstore/src/set_disk.rs
+16 -22
View File
@@ -83,7 +83,7 @@ use rustfs_filemeta::{
headers::{AMZ_OBJECT_TAGGING, AMZ_STORAGE_CLASS},
merge_file_meta_versions,
};
use rustfs_lock::{NamespaceLockManager, NsLockMap};
use rustfs_lock::NamespaceLockManager;
use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem};
use rustfs_rio::{EtagResolvable, HashReader, TryGetIndex as _, WarpReader};
use rustfs_utils::{
@@ -123,9 +123,8 @@ pub const MAX_PARTS_COUNT: usize = 10000;
#[derive(Clone, Debug)]
pub struct SetDisks {
pub lockers: Vec<Arc<rustfs_lock::NamespaceLock>>,
pub namespace_lock: Arc<rustfs_lock::NamespaceLock>,
pub locker_owner: String,
pub ns_mutex: Arc<NsLockMap>,
pub disks: Arc<RwLock<Vec<Option<DiskStore>>>>,
pub set_endpoints: Vec<Endpoint>,
pub set_drive_count: usize,
@@ -138,9 +137,8 @@ pub struct SetDisks {
impl SetDisks {
#[allow(clippy::too_many_arguments)]
pub async fn new(
lockers: Vec<Arc<rustfs_lock::NamespaceLock>>,
namespace_lock: Arc<rustfs_lock::NamespaceLock>,
locker_owner: String,
ns_mutex: Arc<NsLockMap>,
disks: Arc<RwLock<Vec<Option<DiskStore>>>>,
set_drive_count: usize,
default_parity_count: usize,
@@ -150,9 +148,8 @@ impl SetDisks {
format: FormatV3,
) -> Arc<Self> {
Arc::new(SetDisks {
lockers,
namespace_lock,
locker_owner,
ns_mutex,
disks,
set_drive_count,
default_parity_count,
@@ -4066,25 +4063,21 @@ impl ObjectIO for SetDisks {
async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result<ObjectInfo> {
let disks = self.disks.read().await;
let mut _ns = None;
if !opts.no_lock {
let paths = vec![object.to_string()];
let ns_lock = self
.ns_mutex
.new_nslock(None)
.await
.map_err(|err| Error::other(err.to_string()))?;
let lock_acquired = ns_lock
.lock_batch(&paths, &self.locker_owner, std::time::Duration::from_secs(5))
.await
.map_err(|err| Error::other(err.to_string()))?;
let lock_acquired = self
.namespace_lock
.lock_batch(
&paths,
&self.locker_owner,
std::time::Duration::from_secs(5),
std::time::Duration::from_secs(10),
)
.await?;
if !lock_acquired {
return Err(Error::other("can not get lock. please retry".to_string()));
}
_ns = Some(ns_lock);
}
let mut user_defined = opts.user_defined.clone();
@@ -4291,9 +4284,10 @@ impl ObjectIO for SetDisks {
self.delete_all(RUSTFS_META_TMP_BUCKET, &tmp_dir).await?;
if let Some(ns_lock) = _ns {
// Release lock if it was acquired
if !opts.no_lock {
let paths = vec![object.to_string()];
if let Err(err) = ns_lock.unlock_batch(&paths, &self.locker_owner).await {
if let Err(err) = self.namespace_lock.unlock_batch(&paths, &self.locker_owner).await {
error!("Failed to unlock object {}: {}", object, err);
}
}
crates/ecstore/src/sets.rs
+8 -30
View File
@@ -40,11 +40,10 @@ use crate::{
store_init::{check_format_erasure_values, get_format_erasure_in_quorum, load_format_erasure_all, save_format_file},
};
use futures::future::join_all;
use futures_util::FutureExt;
use http::HeaderMap;
use rustfs_common::globals::GLOBAL_Local_Node_Name;
use rustfs_filemeta::FileInfo;
use rustfs_lock::NamespaceLock;
use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem};
use rustfs_utils::{crc_hash, path::path_join_buf, sip_hash};
use tokio::sync::RwLock;
@@ -56,12 +55,13 @@ use tokio::time::Duration;
use tracing::warn;
use tracing::{error, info};
use crate::lock_utils::create_unique_clients;
#[derive(Debug, Clone)]
pub struct Sets {
pub id: Uuid,
// pub sets: Vec<Objects>,
// pub disk_set: Vec<Vec<Option<DiskStore>>>, // [set_count_idx][set_drive_count_idx] = disk_idx
pub lockers: Vec<Vec<Arc<NamespaceLock>>>,
pub disk_set: Vec<Arc<SetDisks>>, // [set_count_idx][set_drive_count_idx] = disk_idx
pub pool_idx: usize,
pub endpoints: PoolEndpoints,
@@ -95,45 +95,24 @@ impl Sets {
let set_drive_count = fm.erasure.sets[0].len();
let mut unique: Vec<Vec<String>> = (0..set_count).map(|_| vec![]).collect();
let mut lockers: Vec<Vec<Arc<NamespaceLock>>> = (0..set_count).map(|_| vec![]).collect();
for (idx, endpoint) in endpoints.endpoints.as_ref().iter().enumerate() {
let set_idx = idx / set_drive_count;
if endpoint.is_local && !unique[set_idx].contains(&"local".to_string()) {
unique[set_idx].push("local".to_string());
let local_manager = rustfs_lock::NsLockMap::new(false, None);
let local_lock = Arc::new(local_manager.new_nslock(None).await.unwrap_or_else(|_| {
// If creation fails, create an empty lock manager
rustfs_lock::NsLockMap::new(false, None)
.new_nslock(None)
.now_or_never()
.unwrap()
.unwrap()
}));
lockers[set_idx].push(local_lock);
}
if !endpoint.is_local {
let host_port = format!("{}:{}", endpoint.url.host_str().unwrap(), endpoint.url.port().unwrap());
if !unique[set_idx].contains(&host_port) {
unique[set_idx].push(host_port);
let dist_manager = rustfs_lock::NsLockMap::new(true, None);
let dist_lock = Arc::new(dist_manager.new_nslock(Some(endpoint.url.clone())).await.unwrap_or_else(|_| {
// If creation fails, create an empty lock manager
rustfs_lock::NsLockMap::new(true, None)
.new_nslock(Some(endpoint.url.clone()))
.now_or_never()
.unwrap()
.unwrap()
}));
lockers[set_idx].push(dist_lock);
}
}
}
let mut disk_set = Vec::with_capacity(set_count);
for (i, locker) in lockers.iter().enumerate().take(set_count) {
for i in 0..set_count {
let mut set_drive = Vec::with_capacity(set_drive_count);
let mut set_endpoints = Vec::with_capacity(set_drive_count);
for j in 0..set_drive_count {
@@ -141,7 +120,6 @@ impl Sets {
let mut disk = disks[idx].clone();
let endpoint = endpoints.endpoints.as_ref()[idx].clone();
// let endpoint = endpoints.endpoints.as_ref().get(idx).cloned();
set_endpoints.push(endpoint);
if disk.is_none() {
@@ -185,12 +163,13 @@ impl Sets {
}
}
// warn!("sets new set_drive {:?}", &set_drive);
let lock_clients = create_unique_clients(&set_endpoints).await?;
let namespace_lock = rustfs_lock::NamespaceLock::with_clients(format!("set-{i}"), lock_clients);
let set_disks = SetDisks::new(
locker.clone(),
Arc::new(namespace_lock),
GLOBAL_Local_Node_Name.read().await.to_string(),
Arc::new(rustfs_lock::NsLockMap::new(is_dist_erasure().await, None)),
Arc::new(RwLock::new(set_drive)),
set_drive_count,
parity_count,
@@ -210,7 +189,6 @@ impl Sets {
id: fm.id,
// sets: todo!(),
disk_set,
lockers,
pool_idx,
endpoints: endpoints.clone(),
format: fm.clone(),
crates/lock/Cargo.toml
+3 -7
View File
@@ -30,6 +30,9 @@ workspace = true
[dependencies]
async-trait.workspace = true
bytes.workspace = true
futures.workspace = true
lazy_static.workspace = true
rustfs-protos.workspace = true
rand.workspace = true
serde.workspace = true
@@ -42,10 +45,3 @@ uuid.workspace = true
thiserror.workspace = true
once_cell.workspace = true
lru.workspace = true
dashmap.workspace = true
[features]
default = []
distributed = []
metrics = []
tracing = []
crates/lock/src/client/local.rs
+180 -47
View File
@@ -37,27 +37,6 @@ impl LocalClient {
pub fn get_lock_map(&self) -> Arc<LocalLockMap> {
crate::get_global_lock_map()
}
/// Convert LockRequest to batch operation
async fn request_to_batch(&self, request: LockRequest) -> Result<bool> {
let lock_map = self.get_lock_map();
let resources = vec![request.resource];
let timeout = request.timeout;
match request.lock_type {
LockType::Exclusive => lock_map.lock_batch(&resources, &request.owner, timeout, None).await,
LockType::Shared => lock_map.rlock_batch(&resources, &request.owner, timeout, None).await,
}
}
/// Convert LockId to resource for release
async fn lock_id_to_batch_release(&self, lock_id: &LockId) -> Result<()> {
let lock_map = self.get_lock_map();
// For simplicity, we'll use the lock_id as resource name
// In a real implementation, you might want to maintain a mapping
let resources = vec![lock_id.as_str().to_string()];
lock_map.unlock_batch(&resources, "unknown").await
}
}
impl Default for LocalClient {
@@ -68,10 +47,10 @@ impl Default for LocalClient {
#[async_trait::async_trait]
impl LockClient for LocalClient {
async fn acquire_exclusive(&self, request: LockRequest) -> Result<LockResponse> {
async fn acquire_exclusive(&self, request: &LockRequest) -> Result<LockResponse> {
let lock_map = self.get_lock_map();
let success = lock_map
.lock_with_ttl_id(&request.resource, &request.owner, request.timeout, None)
.lock_with_ttl_id(request)
.await
.map_err(|e| crate::error::LockError::internal(format!("Lock acquisition failed: {e}")))?;
if success {
@@ -82,7 +61,7 @@ impl LockClient for LocalClient {
status: crate::types::LockStatus::Acquired,
owner: request.owner.clone(),
acquired_at: std::time::SystemTime::now(),
expires_at: std::time::SystemTime::now() + request.timeout,
expires_at: std::time::SystemTime::now() + request.ttl,
last_refreshed: std::time::SystemTime::now(),
metadata: request.metadata.clone(),
priority: request.priority,
@@ -94,10 +73,10 @@ impl LockClient for LocalClient {
}
}
async fn acquire_shared(&self, request: LockRequest) -> Result<LockResponse> {
async fn acquire_shared(&self, request: &LockRequest) -> Result<LockResponse> {
let lock_map = self.get_lock_map();
let success = lock_map
.rlock_with_ttl_id(&request.resource, &request.owner, request.timeout, None)
.rlock_with_ttl_id(request)
.await
.map_err(|e| crate::error::LockError::internal(format!("Shared lock acquisition failed: {e}")))?;
if success {
@@ -108,7 +87,7 @@ impl LockClient for LocalClient {
status: crate::types::LockStatus::Acquired,
owner: request.owner.clone(),
acquired_at: std::time::SystemTime::now(),
expires_at: std::time::SystemTime::now() + request.timeout,
expires_at: std::time::SystemTime::now() + request.ttl,
last_refreshed: std::time::SystemTime::now(),
metadata: request.metadata.clone(),
priority: request.priority,
@@ -122,11 +101,19 @@ impl LockClient for LocalClient {
async fn release(&self, lock_id: &LockId) -> Result<bool> {
let lock_map = self.get_lock_map();
lock_map
.unlock_by_id(lock_id)
.await
.map_err(|e| crate::error::LockError::internal(format!("Release failed: {e}")))?;
Ok(true)
// Try to release the lock directly by ID
match lock_map.unlock_by_id(lock_id).await {
Ok(()) => Ok(true),
Err(e) if e.kind() == std::io::ErrorKind::NotFound => {
// Try as read lock if exclusive unlock failed
match lock_map.runlock_by_id(lock_id).await {
Ok(()) => Ok(true),
Err(_) => Err(crate::error::LockError::internal("Lock ID not found".to_string())),
}
}
Err(e) => Err(crate::error::LockError::internal(format!("Release lock failed: {e}"))),
}
}
async fn refresh(&self, _lock_id: &LockId) -> Result<bool> {
@@ -140,15 +127,34 @@ impl LockClient for LocalClient {
async fn check_status(&self, lock_id: &LockId) -> Result<Option<LockInfo>> {
let lock_map = self.get_lock_map();
if let Some((resource, owner)) = lock_map.lockid_map.get(lock_id).map(|v| v.clone()) {
let is_locked = lock_map.is_locked(&resource).await;
if is_locked {
// Check if the lock exists in our locks map
let locks_guard = lock_map.locks.read().await;
if let Some(entry) = locks_guard.get(lock_id) {
let entry_guard = entry.read().await;
// Determine lock type and owner based on the entry
if let Some(owner) = &entry_guard.writer {
Ok(Some(LockInfo {
id: lock_id.clone(),
resource,
lock_type: LockType::Exclusive, // 这里可进一步完善
resource: lock_id.resource.clone(),
lock_type: crate::types::LockType::Exclusive,
status: crate::types::LockStatus::Acquired,
owner,
owner: owner.clone(),
acquired_at: std::time::SystemTime::now(),
expires_at: std::time::SystemTime::now() + std::time::Duration::from_secs(30),
last_refreshed: std::time::SystemTime::now(),
metadata: LockMetadata::default(),
priority: LockPriority::Normal,
wait_start_time: None,
}))
} else if !entry_guard.readers.is_empty() {
Ok(Some(LockInfo {
id: lock_id.clone(),
resource: lock_id.resource.clone(),
lock_type: crate::types::LockType::Shared,
status: crate::types::LockStatus::Acquired,
owner: entry_guard.readers.iter().next().map(|(k, _)| k.clone()).unwrap_or_default(),
acquired_at: std::time::SystemTime::now(),
expires_at: std::time::SystemTime::now() + std::time::Duration::from_secs(30),
last_refreshed: std::time::SystemTime::now(),
@@ -189,31 +195,44 @@ mod tests {
#[tokio::test]
async fn test_local_client_acquire_exclusive() {
let client = LocalClient::new();
let request =
LockRequest::new("test-resource", LockType::Exclusive, "test-owner").with_timeout(std::time::Duration::from_secs(30));
let resource_name = format!("test-resource-exclusive-{}", uuid::Uuid::new_v4());
let request = LockRequest::new(&resource_name, LockType::Exclusive, "test-owner")
.with_acquire_timeout(std::time::Duration::from_secs(30));
let response = client.acquire_exclusive(request).await.unwrap();
let response = client.acquire_exclusive(&request).await.unwrap();
assert!(response.is_success());
// Clean up
if let Some(lock_info) = response.lock_info() {
let _ = client.release(&lock_info.id).await;
}
}
#[tokio::test]
async fn test_local_client_acquire_shared() {
let client = LocalClient::new();
let request =
LockRequest::new("test-resource", LockType::Shared, "test-owner").with_timeout(std::time::Duration::from_secs(30));
let resource_name = format!("test-resource-shared-{}", uuid::Uuid::new_v4());
let request = LockRequest::new(&resource_name, LockType::Shared, "test-owner")
.with_acquire_timeout(std::time::Duration::from_secs(30));
let response = client.acquire_shared(request).await.unwrap();
let response = client.acquire_shared(&request).await.unwrap();
assert!(response.is_success());
// Clean up
if let Some(lock_info) = response.lock_info() {
let _ = client.release(&lock_info.id).await;
}
}
#[tokio::test]
async fn test_local_client_release() {
let client = LocalClient::new();
let resource_name = format!("test-resource-release-{}", uuid::Uuid::new_v4());
// First acquire a lock
let request =
LockRequest::new("test-resource", LockType::Exclusive, "test-owner").with_timeout(std::time::Duration::from_secs(30));
let response = client.acquire_exclusive(request).await.unwrap();
let request = LockRequest::new(&resource_name, LockType::Exclusive, "test-owner")
.with_acquire_timeout(std::time::Duration::from_secs(30));
let response = client.acquire_exclusive(&request).await.unwrap();
assert!(response.is_success());
// Get the lock ID from the response
@@ -230,4 +249,118 @@ mod tests {
let client = LocalClient::new();
assert!(client.is_local().await);
}
#[tokio::test]
async fn test_local_client_read_write_lock_exclusion() {
let client = LocalClient::new();
let resource_name = format!("test-resource-exclusion-{}", uuid::Uuid::new_v4());
// First, acquire an exclusive lock
let exclusive_request = LockRequest::new(&resource_name, LockType::Exclusive, "exclusive-owner")
.with_acquire_timeout(std::time::Duration::from_millis(10));
let exclusive_response = client.acquire_exclusive(&exclusive_request).await.unwrap();
assert!(exclusive_response.is_success());
// Try to acquire a shared lock on the same resource - should fail
let shared_request = LockRequest::new(&resource_name, LockType::Shared, "shared-owner")
.with_acquire_timeout(std::time::Duration::from_millis(10));
let shared_response = client.acquire_shared(&shared_request).await.unwrap();
assert!(!shared_response.is_success(), "Shared lock should fail when exclusive lock exists");
// Clean up exclusive lock
if let Some(exclusive_info) = exclusive_response.lock_info() {
let _ = client.release(&exclusive_info.id).await;
}
// Now shared lock should succeed
let shared_request2 = LockRequest::new(&resource_name, LockType::Shared, "shared-owner")
.with_acquire_timeout(std::time::Duration::from_millis(10));
let shared_response2 = client.acquire_shared(&shared_request2).await.unwrap();
assert!(
shared_response2.is_success(),
"Shared lock should succeed after exclusive lock is released"
);
// Clean up
if let Some(shared_info) = shared_response2.lock_info() {
let _ = client.release(&shared_info.id).await;
}
}
#[tokio::test]
async fn test_local_client_read_write_lock_distinction() {
let client = LocalClient::new();
let resource_name = format!("test-resource-rw-{}", uuid::Uuid::new_v4());
// Test exclusive lock
let exclusive_request = LockRequest::new(&resource_name, LockType::Exclusive, "exclusive-owner")
.with_acquire_timeout(std::time::Duration::from_secs(30));
let exclusive_response = client.acquire_exclusive(&exclusive_request).await.unwrap();
assert!(exclusive_response.is_success());
if let Some(exclusive_info) = exclusive_response.lock_info() {
assert_eq!(exclusive_info.lock_type, LockType::Exclusive);
// Check status should return correct lock type
let status = client.check_status(&exclusive_info.id).await.unwrap();
assert!(status.is_some());
assert_eq!(status.unwrap().lock_type, LockType::Exclusive);
// Release exclusive lock
let result = client.release(&exclusive_info.id).await.unwrap();
assert!(result);
}
// Test shared lock
let shared_request = LockRequest::new(&resource_name, LockType::Shared, "shared-owner")
.with_acquire_timeout(std::time::Duration::from_secs(30));
let shared_response = client.acquire_shared(&shared_request).await.unwrap();
assert!(shared_response.is_success());
if let Some(shared_info) = shared_response.lock_info() {
assert_eq!(shared_info.lock_type, LockType::Shared);
// Check status should return correct lock type
let status = client.check_status(&shared_info.id).await.unwrap();
assert!(status.is_some());
assert_eq!(status.unwrap().lock_type, LockType::Shared);
// Release shared lock
let result = client.release(&shared_info.id).await.unwrap();
assert!(result);
}
}
#[tokio::test]
async fn test_multiple_local_clients_exclusive_mutex() {
let client1 = LocalClient::new();
let client2 = LocalClient::new();
let resource_name = format!("test-multi-client-mutex-{}", uuid::Uuid::new_v4());
// client1 acquire exclusive lock
let req1 = LockRequest::new(&resource_name, LockType::Exclusive, "owner1")
.with_acquire_timeout(std::time::Duration::from_millis(50));
let resp1 = client1.acquire_exclusive(&req1).await.unwrap();
assert!(resp1.is_success(), "client1 should acquire exclusive lock");
// client2 try to acquire exclusive lock, should fail
let req2 = LockRequest::new(&resource_name, LockType::Exclusive, "owner2")
.with_acquire_timeout(std::time::Duration::from_millis(50));
let resp2 = client2.acquire_exclusive(&req2).await.unwrap();
assert!(!resp2.is_success(), "client2 should not acquire exclusive lock while client1 holds it");
// client1 release lock
if let Some(lock_info) = resp1.lock_info() {
let _ = client1.release(&lock_info.id).await;
}
// client2 try again, should succeed
let resp3 = client2.acquire_exclusive(&req2).await.unwrap();
assert!(resp3.is_success(), "client2 should acquire exclusive lock after client1 releases it");
// clean up
if let Some(lock_info) = resp3.lock_info() {
let _ = client2.release(&lock_info.id).await;
}
}
}
crates/lock/src/client/mod.rs
+4 -4
View File
@@ -27,13 +27,13 @@ use crate::{
#[async_trait]
pub trait LockClient: Send + Sync + std::fmt::Debug {
/// Acquire exclusive lock
async fn acquire_exclusive(&self, request: LockRequest) -> Result<LockResponse>;
async fn acquire_exclusive(&self, request: &LockRequest) -> Result<LockResponse>;
/// Acquire shared lock
async fn acquire_shared(&self, request: LockRequest) -> Result<LockResponse>;
async fn acquire_shared(&self, request: &LockRequest) -> Result<LockResponse>;
/// Acquire lock (generic method)
async fn acquire_lock(&self, request: LockRequest) -> Result<LockResponse> {
async fn acquire_lock(&self, request: &LockRequest) -> Result<LockResponse> {
match request.lock_type {
crate::types::LockType::Exclusive => self.acquire_exclusive(request).await,
crate::types::LockType::Shared => self.acquire_shared(request).await,
@@ -101,7 +101,7 @@ mod tests {
let request = crate::types::LockRequest::new("test-resource", LockType::Exclusive, "test-owner");
// Test lock acquisition
let response = client.acquire_exclusive(request).await;
let response = client.acquire_exclusive(&request).await;
assert!(response.is_ok());
if let Ok(response) = response {
crates/lock/src/client/remote.rs
+269 -95
View File
@@ -13,8 +13,13 @@
// limitations under the License.
use async_trait::async_trait;
use rustfs_protos::{node_service_time_out_client, proto_gen::node_service::GenerallyLockRequest};
use serde::{Deserialize, Serialize};
use rustfs_protos::{
node_service_time_out_client,
proto_gen::node_service::{GenerallyLockRequest, PingRequest},
};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use tonic::Request;
use tracing::info;
@@ -25,154 +30,220 @@ use crate::{
use super::LockClient;
/// RPC lock arguments for gRPC communication
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LockArgs {
pub uid: String,
pub resources: Vec<String>,
pub owner: String,
pub source: String,
pub quorum: u32,
}
impl LockArgs {
fn from_request(request: &LockRequest, _is_shared: bool) -> Self {
Self {
uid: request.metadata.operation_id.clone().unwrap_or_default(),
resources: vec![request.resource.clone()],
owner: request.owner.clone(),
source: "remote".to_string(),
quorum: 1,
}
}
fn from_lock_id(lock_id: &LockId) -> Self {
Self {
uid: lock_id.as_str().to_string(),
resources: vec![lock_id.as_str().to_string()],
owner: "remote".to_string(),
source: "remote".to_string(),
quorum: 1,
}
}
}
/// Remote lock client implementation
#[derive(Debug, Clone)]
#[derive(Debug)]
pub struct RemoteClient {
addr: String,
// Track active locks with their original owner information
active_locks: Arc<RwLock<HashMap<LockId, String>>>, // lock_id -> owner
}
impl Clone for RemoteClient {
fn clone(&self) -> Self {
Self {
addr: self.addr.clone(),
active_locks: self.active_locks.clone(),
}
}
}
impl RemoteClient {
pub fn new(endpoint: String) -> Self {
Self { addr: endpoint }
Self {
addr: endpoint,
active_locks: Arc::new(RwLock::new(HashMap::new())),
}
}
pub fn from_url(url: url::Url) -> Self {
Self { addr: url.to_string() }
Self {
addr: url.to_string(),
active_locks: Arc::new(RwLock::new(HashMap::new())),
}
}
/// Create a minimal LockRequest for unlock operations
fn create_unlock_request(&self, lock_id: &LockId, owner: &str) -> LockRequest {
LockRequest {
lock_id: lock_id.clone(),
resource: lock_id.resource.clone(),
lock_type: crate::types::LockType::Exclusive, // Type doesn't matter for unlock
owner: owner.to_string(),
acquire_timeout: std::time::Duration::from_secs(30),
ttl: std::time::Duration::from_secs(300),
metadata: crate::types::LockMetadata::default(),
priority: crate::types::LockPriority::Normal,
deadlock_detection: false,
}
}
}
#[async_trait]
impl LockClient for RemoteClient {
async fn acquire_exclusive(&self, request: LockRequest) -> Result<LockResponse> {
async fn acquire_exclusive(&self, request: &LockRequest) -> Result<LockResponse> {
info!("remote acquire_exclusive for {}", request.resource);
let args = LockArgs::from_request(&request, false);
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| LockError::internal(format!("can not get client, err: {err}")))?;
let req = Request::new(GenerallyLockRequest {
args: serde_json::to_string(&args).map_err(|e| LockError::internal(format!("Failed to serialize args: {e}")))?,
args: serde_json::to_string(&request)
.map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
});
let resp = client
.lock(req)
.await
.map_err(|e| LockError::internal(e.to_string()))?
.into_inner();
// Check for explicit error first
if let Some(error_info) = resp.error_info {
return Err(LockError::internal(error_info));
}
Ok(LockResponse::success(
LockInfo {
id: LockId::new_deterministic(&request.resource),
resource: request.resource,
lock_type: request.lock_type,
status: crate::types::LockStatus::Acquired,
owner: request.owner,
acquired_at: std::time::SystemTime::now(),
expires_at: std::time::SystemTime::now() + request.timeout,
last_refreshed: std::time::SystemTime::now(),
metadata: request.metadata,
priority: request.priority,
wait_start_time: None,
},
std::time::Duration::ZERO,
))
// Check if the lock acquisition was successful
if resp.success {
// Save the lock information for later release
let mut locks = self.active_locks.write().await;
locks.insert(request.lock_id.clone(), request.owner.clone());
Ok(LockResponse::success(
LockInfo {
id: request.lock_id.clone(),
resource: request.resource.clone(),
lock_type: request.lock_type,
status: crate::types::LockStatus::Acquired,
owner: request.owner.clone(),
acquired_at: std::time::SystemTime::now(),
expires_at: std::time::SystemTime::now() + request.ttl,
last_refreshed: std::time::SystemTime::now(),
metadata: request.metadata.clone(),
priority: request.priority,
wait_start_time: None,
},
std::time::Duration::ZERO,
))
} else {
// Lock acquisition failed
Ok(LockResponse::failure(
"Lock acquisition failed on remote server".to_string(),
std::time::Duration::ZERO,
))
}
}
async fn acquire_shared(&self, request: LockRequest) -> Result<LockResponse> {
async fn acquire_shared(&self, request: &LockRequest) -> Result<LockResponse> {
info!("remote acquire_shared for {}", request.resource);
let args = LockArgs::from_request(&request, true);
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| LockError::internal(format!("can not get client, err: {err}")))?;
let req = Request::new(GenerallyLockRequest {
args: serde_json::to_string(&args).map_err(|e| LockError::internal(format!("Failed to serialize args: {e}")))?,
args: serde_json::to_string(&request)
.map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
});
let resp = client
.r_lock(req)
.await
.map_err(|e| LockError::internal(e.to_string()))?
.into_inner();
// Check for explicit error first
if let Some(error_info) = resp.error_info {
return Err(LockError::internal(error_info));
}
Ok(LockResponse::success(
LockInfo {
id: LockId::new_deterministic(&request.resource),
resource: request.resource,
lock_type: request.lock_type,
status: crate::types::LockStatus::Acquired,
owner: request.owner,
acquired_at: std::time::SystemTime::now(),
expires_at: std::time::SystemTime::now() + request.timeout,
last_refreshed: std::time::SystemTime::now(),
metadata: request.metadata,
priority: request.priority,
wait_start_time: None,
},
std::time::Duration::ZERO,
))
// Check if the lock acquisition was successful
if resp.success {
// Save the lock information for later release
let mut locks = self.active_locks.write().await;
locks.insert(request.lock_id.clone(), request.owner.clone());
Ok(LockResponse::success(
LockInfo {
id: request.lock_id.clone(),
resource: request.resource.clone(),
lock_type: request.lock_type,
status: crate::types::LockStatus::Acquired,
owner: request.owner.clone(),
acquired_at: std::time::SystemTime::now(),
expires_at: std::time::SystemTime::now() + request.ttl,
last_refreshed: std::time::SystemTime::now(),
metadata: request.metadata.clone(),
priority: request.priority,
wait_start_time: None,
},
std::time::Duration::ZERO,
))
} else {
// Lock acquisition failed
Ok(LockResponse::failure(
"Shared lock acquisition failed on remote server".to_string(),
std::time::Duration::ZERO,
))
}
}
async fn release(&self, lock_id: &LockId) -> Result<bool> {
info!("remote release for {}", lock_id);
let args = LockArgs::from_lock_id(lock_id);
// Get the original owner for this lock
let owner = {
let locks = self.active_locks.read().await;
locks.get(lock_id).cloned().unwrap_or_else(|| "remote".to_string())
};
let unlock_request = self.create_unlock_request(lock_id, &owner);
let request_string = serde_json::to_string(&unlock_request)
.map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?;
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| LockError::internal(format!("can not get client, err: {err}")))?;
// Try UnLock first (for exclusive locks)
let req = Request::new(GenerallyLockRequest {
args: serde_json::to_string(&args).map_err(|e| LockError::internal(format!("Failed to serialize args: {e}")))?,
args: request_string.clone(),
});
let resp = client
.un_lock(req)
.await
.map_err(|e| LockError::internal(e.to_string()))?
.into_inner();
if let Some(error_info) = resp.error_info {
return Err(LockError::internal(error_info));
let resp = client.un_lock(req).await;
let success = if resp.is_err() {
// If that fails, try RUnLock (for shared locks)
let req = Request::new(GenerallyLockRequest { args: request_string });
let resp = client
.r_un_lock(req)
.await
.map_err(|e| LockError::internal(e.to_string()))?
.into_inner();
if let Some(error_info) = resp.error_info {
return Err(LockError::internal(error_info));
}
resp.success
} else {
let resp = resp.map_err(|e| LockError::internal(e.to_string()))?.into_inner();
if let Some(error_info) = resp.error_info {
return Err(LockError::internal(error_info));
}
resp.success
};
// Remove the lock from our tracking if successful
if success {
let mut locks = self.active_locks.write().await;
locks.remove(lock_id);
}
Ok(resp.success)
Ok(success)
}
async fn refresh(&self, lock_id: &LockId) -> Result<bool> {
info!("remote refresh for {}", lock_id);
let args = LockArgs::from_lock_id(lock_id);
let refresh_request = self.create_unlock_request(lock_id, "remote");
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| LockError::internal(format!("can not get client, err: {err}")))?;
let req = Request::new(GenerallyLockRequest {
args: serde_json::to_string(&args).map_err(|e| LockError::internal(format!("Failed to serialize args: {e}")))?,
args: serde_json::to_string(&refresh_request)
.map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
});
let resp = client
.refresh(req)
@@ -187,12 +258,13 @@ impl LockClient for RemoteClient {
async fn force_release(&self, lock_id: &LockId) -> Result<bool> {
info!("remote force_release for {}", lock_id);
let args = LockArgs::from_lock_id(lock_id);
let force_request = self.create_unlock_request(lock_id, "remote");
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| LockError::internal(format!("can not get client, err: {err}")))?;
let req = Request::new(GenerallyLockRequest {
args: serde_json::to_string(&args).map_err(|e| LockError::internal(format!("Failed to serialize args: {e}")))?,
args: serde_json::to_string(&force_request)
.map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
});
let resp = client
.force_un_lock(req)
@@ -205,14 +277,93 @@ impl LockClient for RemoteClient {
Ok(resp.success)
}
async fn check_status(&self, _lock_id: &LockId) -> Result<Option<LockInfo>> {
// TODO: Implement remote status query
Ok(None)
async fn check_status(&self, lock_id: &LockId) -> Result<Option<LockInfo>> {
info!("remote check_status for {}", lock_id);
// Since there's no direct status query in the gRPC service,
// we attempt a non-blocking lock acquisition to check if the resource is available
let status_request = self.create_unlock_request(lock_id, "remote");
let mut client = node_service_time_out_client(&self.addr)
.await
.map_err(|err| LockError::internal(format!("can not get client, err: {err}")))?;
// Try to acquire a very short-lived lock to test availability
let req = Request::new(GenerallyLockRequest {
args: serde_json::to_string(&status_request)
.map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
});
// Try exclusive lock first with very short timeout
let resp = client.lock(req).await;
match resp {
Ok(response) => {
let resp = response.into_inner();
if resp.success {
// If we successfully acquired the lock, the resource was free
// Immediately release it
let release_req = Request::new(GenerallyLockRequest {
args: serde_json::to_string(&status_request)
.map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?,
});
let _ = client.un_lock(release_req).await; // Best effort release
// Return None since no one was holding the lock
Ok(None)
} else {
// Lock acquisition failed, meaning someone is holding it
// We can't determine the exact details remotely, so return a generic status
Ok(Some(LockInfo {
id: lock_id.clone(),
resource: lock_id.as_str().to_string(),
lock_type: crate::types::LockType::Exclusive, // We can't know the exact type
status: crate::types::LockStatus::Acquired,
owner: "unknown".to_string(), // Remote client can't determine owner
acquired_at: std::time::SystemTime::now(),
expires_at: std::time::SystemTime::now() + std::time::Duration::from_secs(3600),
last_refreshed: std::time::SystemTime::now(),
metadata: crate::types::LockMetadata::default(),
priority: crate::types::LockPriority::Normal,
wait_start_time: None,
}))
}
}
Err(_) => {
// Communication error or lock is held
Ok(Some(LockInfo {
id: lock_id.clone(),
resource: lock_id.as_str().to_string(),
lock_type: crate::types::LockType::Exclusive,
status: crate::types::LockStatus::Acquired,
owner: "unknown".to_string(),
acquired_at: std::time::SystemTime::now(),
expires_at: std::time::SystemTime::now() + std::time::Duration::from_secs(3600),
last_refreshed: std::time::SystemTime::now(),
metadata: crate::types::LockMetadata::default(),
priority: crate::types::LockPriority::Normal,
wait_start_time: None,
}))
}
}
}
async fn get_stats(&self) -> Result<LockStats> {
// TODO: Implement remote statistics
Ok(LockStats::default())
info!("remote get_stats from {}", self.addr);
// Since there's no direct statistics endpoint in the gRPC service,
// we return basic stats indicating this is a remote client
let stats = LockStats {
last_updated: std::time::SystemTime::now(),
..Default::default()
};
// We could potentially enhance this by:
// 1. Keeping local counters of operations performed
// 2. Adding a stats gRPC method to the service
// 3. Querying server health endpoints
// For now, return minimal stats indicating remote connectivity
Ok(stats)
}
async fn close(&self) -> Result<()> {
@@ -220,7 +371,30 @@ impl LockClient for RemoteClient {
}
async fn is_online(&self) -> bool {
true
// Use Ping interface to test if remote service is online
let mut client = match node_service_time_out_client(&self.addr).await {
Ok(client) => client,
Err(_) => {
info!("remote client {} connection failed", self.addr);
return false;
}
};
let ping_req = Request::new(PingRequest {
version: 1,
body: bytes::Bytes::new(),
});
match client.ping(ping_req).await {
Ok(_) => {
info!("remote client {} is online", self.addr);
true
}
Err(_) => {
info!("remote client {} ping failed", self.addr);
false
}
}
}
async fn is_local(&self) -> bool {
crates/lock/src/config.rs
-201
View File
@@ -1,201 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use serde::{Deserialize, Serialize};
use std::time::Duration;
/// Lock system configuration
#[derive(Debug, Clone, Serialize, Deserialize, Default)]
pub struct LockConfig {
/// Whether distributed locking is enabled
pub distributed_enabled: bool,
/// Local lock configuration
pub local: LocalLockConfig,
/// Distributed lock configuration
pub distributed: DistributedLockConfig,
/// Network configuration
pub network: NetworkConfig,
}
/// Local lock configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct LocalLockConfig {
/// Default lock timeout
pub default_timeout: Duration,
/// Default lock expiration time
pub default_expiration: Duration,
/// Maximum number of locks per resource
pub max_locks_per_resource: usize,
}
/// Distributed lock configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct DistributedLockConfig {
/// Total number of nodes in the cluster
pub total_nodes: usize,
/// Number of nodes that can fail (tolerance)
pub tolerance: usize,
/// Lock acquisition timeout
pub acquisition_timeout: Duration,
/// Lock refresh interval
pub refresh_interval: Duration,
/// Lock expiration time
pub expiration_time: Duration,
/// Retry interval for failed operations
pub retry_interval: Duration,
/// Maximum number of retry attempts
pub max_retries: usize,
}
/// Network configuration
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct NetworkConfig {
/// Connection timeout
pub connection_timeout: Duration,
/// Request timeout
pub request_timeout: Duration,
/// Keep-alive interval
pub keep_alive_interval: Duration,
/// Maximum connection pool size
pub max_connections: usize,
}
impl Default for LocalLockConfig {
fn default() -> Self {
Self {
default_timeout: Duration::from_secs(30),
default_expiration: Duration::from_secs(60),
max_locks_per_resource: 1000,
}
}
}
impl Default for DistributedLockConfig {
fn default() -> Self {
Self {
total_nodes: 3,
tolerance: 1,
acquisition_timeout: Duration::from_secs(30),
refresh_interval: Duration::from_secs(10),
expiration_time: Duration::from_secs(60),
retry_interval: Duration::from_millis(250),
max_retries: 10,
}
}
}
impl Default for NetworkConfig {
fn default() -> Self {
Self {
connection_timeout: Duration::from_secs(5),
request_timeout: Duration::from_secs(30),
keep_alive_interval: Duration::from_secs(30),
max_connections: 100,
}
}
}
impl LockConfig {
/// Create new lock configuration
pub fn new() -> Self {
Self::default()
}
/// Create distributed lock configuration
pub fn distributed(total_nodes: usize, tolerance: usize) -> Self {
Self {
distributed_enabled: true,
distributed: DistributedLockConfig {
total_nodes,
tolerance,
..Default::default()
},
..Default::default()
}
}
/// Create local-only lock configuration
pub fn local() -> Self {
Self {
distributed_enabled: false,
..Default::default()
}
}
/// Check if distributed locking is enabled
pub fn is_distributed(&self) -> bool {
self.distributed_enabled
}
/// Get quorum size for distributed locks
pub fn get_quorum_size(&self) -> usize {
self.distributed.total_nodes - self.distributed.tolerance
}
/// Check if quorum configuration is valid
pub fn is_quorum_valid(&self) -> bool {
self.distributed.tolerance < self.distributed.total_nodes
}
/// Get effective timeout
pub fn get_effective_timeout(&self, timeout: Option<Duration>) -> Duration {
timeout.unwrap_or(self.local.default_timeout)
}
/// Get effective expiration
pub fn get_effective_expiration(&self, expiration: Option<Duration>) -> Duration {
expiration.unwrap_or(self.local.default_expiration)
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn test_lock_config_default() {
let config = LockConfig::default();
assert!(!config.distributed_enabled);
assert_eq!(config.local.default_timeout, Duration::from_secs(30));
}
#[test]
fn test_lock_config_distributed() {
let config = LockConfig::distributed(5, 2);
assert!(config.distributed_enabled);
assert_eq!(config.distributed.total_nodes, 5);
assert_eq!(config.distributed.tolerance, 2);
assert_eq!(config.get_quorum_size(), 3);
}
#[test]
fn test_lock_config_local() {
let config = LockConfig::local();
assert!(!config.distributed_enabled);
}
#[test]
fn test_effective_timeout() {
let config = LockConfig::default();
assert_eq!(config.get_effective_timeout(None), Duration::from_secs(30));
assert_eq!(config.get_effective_timeout(Some(Duration::from_secs(10))), Duration::from_secs(10));
}
#[test]
fn test_effective_expiration() {
let config = LockConfig::default();
assert_eq!(config.get_effective_expiration(None), Duration::from_secs(60));
assert_eq!(config.get_effective_expiration(Some(Duration::from_secs(30))), Duration::from_secs(30));
}
}
crates/lock/src/distributed.rs
-640
View File
@@ -1,640 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime};
use tokio::sync::{Mutex, RwLock};
use tokio::time::{interval, timeout};
use uuid::Uuid;
use crate::{
client::LockClient,
error::{LockError, Result},
types::{LockId, LockInfo, LockPriority, LockRequest, LockResponse, LockStats, LockStatus, LockType},
};
/// Quorum configuration for distributed locking
#[derive(Debug, Clone)]
pub struct QuorumConfig {
/// Total number of nodes in the cluster
pub total_nodes: usize,
/// Number of nodes that can fail (tolerance)
pub tolerance: usize,
/// Minimum number of nodes required for quorum
pub quorum: usize,
/// Lock acquisition timeout
pub acquisition_timeout: Duration,
/// Lock refresh interval
pub refresh_interval: Duration,
/// Lock expiration time
pub expiration_time: Duration,
}
impl QuorumConfig {
/// Create new quorum configuration
pub fn new(total_nodes: usize, tolerance: usize) -> Self {
let quorum = total_nodes - tolerance;
Self {
total_nodes,
tolerance,
quorum,
acquisition_timeout: Duration::from_secs(30),
refresh_interval: Duration::from_secs(10),
expiration_time: Duration::from_secs(60),
}
}
/// Check if quorum is valid
pub fn is_valid(&self) -> bool {
self.quorum > 0 && self.quorum <= self.total_nodes && self.tolerance < self.total_nodes
}
}
/// Distributed lock entry
#[derive(Debug)]
pub struct DistributedLockEntry {
/// Lock ID
pub lock_id: LockId,
/// Resource being locked
pub resource: String,
/// Lock type
pub lock_type: LockType,
/// Lock owner
pub owner: String,
/// Lock priority
pub priority: LockPriority,
/// Lock acquisition time
pub acquired_at: Instant,
/// Lock expiration time
pub expires_at: Instant,
/// Nodes that hold this lock
pub holders: Vec<String>,
/// Lock refresh task handle
pub refresh_handle: Option<tokio::task::JoinHandle<()>>,
}
impl DistributedLockEntry {
/// Create new distributed lock entry
pub fn new(
lock_id: LockId,
resource: String,
lock_type: LockType,
owner: String,
priority: LockPriority,
expiration_time: Duration,
) -> Self {
let now = Instant::now();
Self {
lock_id,
resource,
lock_type,
owner,
priority,
acquired_at: now,
expires_at: now + expiration_time,
holders: Vec::new(),
refresh_handle: None,
}
}
/// Check if lock has expired
pub fn has_expired(&self) -> bool {
Instant::now() >= self.expires_at
}
/// Extend lock expiration
pub fn extend(&mut self, duration: Duration) {
self.expires_at = Instant::now() + duration;
}
/// Get remaining time until expiration
pub fn remaining_time(&self) -> Duration {
if self.has_expired() {
Duration::ZERO
} else {
self.expires_at - Instant::now()
}
}
}
/// Distributed lock manager
#[derive(Debug)]
pub struct DistributedLockManager {
/// Quorum configuration
config: QuorumConfig,
/// Lock clients for each node
clients: Arc<RwLock<HashMap<String, Arc<dyn LockClient>>>>,
/// Active locks
locks: Arc<RwLock<HashMap<String, DistributedLockEntry>>>,
/// Node ID
node_id: String,
/// Statistics
stats: Arc<Mutex<LockStats>>,
}
impl DistributedLockManager {
/// Create new distributed lock manager
pub fn new(config: QuorumConfig, node_id: String) -> Self {
Self {
config,
clients: Arc::new(RwLock::new(HashMap::new())),
locks: Arc::new(RwLock::new(HashMap::new())),
node_id,
stats: Arc::new(Mutex::new(LockStats::default())),
}
}
/// Add lock client for a node
pub async fn add_client(&self, node_id: String, client: Arc<dyn LockClient>) {
let mut clients = self.clients.write().await;
clients.insert(node_id, client);
}
/// Remove lock client for a node
pub async fn remove_client(&self, node_id: &str) {
let mut clients = self.clients.write().await;
clients.remove(node_id);
}
/// Acquire distributed lock
pub async fn acquire_lock(&self, request: LockRequest) -> Result<LockResponse> {
let resource_key = self.get_resource_key(&request.resource);
// Check if we already hold this lock
{
let locks = self.locks.read().await;
if let Some(lock) = locks.get(&resource_key) {
if lock.owner == request.owner && !lock.has_expired() {
return Ok(LockResponse::success(
LockInfo {
id: lock.lock_id.clone(),
resource: request.resource.clone(),
lock_type: request.lock_type,
status: LockStatus::Acquired,
owner: request.owner.clone(),
acquired_at: SystemTime::now(),
expires_at: SystemTime::now() + lock.remaining_time(),
last_refreshed: SystemTime::now(),
metadata: request.metadata.clone(),
priority: request.priority,
wait_start_time: None,
},
Duration::ZERO,
));
}
}
}
// Try to acquire lock directly
match self.try_acquire_lock(&request).await {
Ok(response) => {
if response.success {
return Ok(response);
}
}
Err(e) => {
tracing::warn!("Direct lock acquisition failed: {}", e);
}
}
// If direct acquisition fails, return timeout error
Err(LockError::timeout("Distributed lock acquisition failed", request.timeout))
}
/// Try to acquire lock directly
async fn try_acquire_lock(&self, request: &LockRequest) -> Result<LockResponse> {
let resource_key = self.get_resource_key(&request.resource);
let clients = self.clients.read().await;
if clients.len() < self.config.quorum {
return Err(LockError::InsufficientNodes {
required: self.config.quorum,
available: clients.len(),
});
}
// Prepare lock request for all nodes
let lock_request = LockRequest {
lock_id: request.lock_id.clone(),
resource: request.resource.clone(),
lock_type: request.lock_type,
owner: request.owner.clone(),
priority: request.priority,
timeout: self.config.acquisition_timeout,
metadata: request.metadata.clone(),
wait_timeout: request.wait_timeout,
deadlock_detection: request.deadlock_detection,
};
// Send lock request to all nodes
let mut responses = Vec::new();
let mut handles = Vec::new();
for (node_id, client) in clients.iter() {
let client = client.clone();
let request = lock_request.clone();
let handle = tokio::spawn(async move { client.acquire_lock(request).await });
handles.push((node_id.clone(), handle));
}
// Collect responses with timeout
for (node_id, handle) in handles {
match timeout(self.config.acquisition_timeout, handle).await {
Ok(Ok(response)) => {
responses.push((node_id, response));
}
Ok(Err(e)) => {
tracing::warn!("Lock request failed for node {}: {}", node_id, e);
}
Err(_) => {
tracing::warn!("Lock request timeout for node {}", node_id);
}
}
}
// Check if we have quorum
let successful_responses = responses
.iter()
.filter(|(_, response)| response.as_ref().map(|r| r.success).unwrap_or(false))
.count();
if successful_responses >= self.config.quorum {
// Create lock entry
let mut lock_entry = DistributedLockEntry::new(
request.lock_id.clone(),
request.resource.clone(),
request.lock_type,
request.owner.clone(),
request.priority,
self.config.expiration_time,
);
// Add successful nodes as holders
for (node_id, _) in responses
.iter()
.filter(|(_, r)| r.as_ref().map(|resp| resp.success).unwrap_or(false))
{
lock_entry.holders.push(node_id.clone());
}
// Start refresh task
let refresh_handle = self.start_refresh_task(&lock_entry).await;
// Store lock entry
{
let mut locks = self.locks.write().await;
lock_entry.refresh_handle = Some(refresh_handle);
locks.insert(resource_key, lock_entry);
}
// Update statistics
self.update_stats(true).await;
Ok(LockResponse::success(
LockInfo {
id: request.lock_id.clone(),
resource: request.resource.clone(),
lock_type: request.lock_type,
status: LockStatus::Acquired,
owner: request.owner.clone(),
acquired_at: SystemTime::now(),
expires_at: SystemTime::now() + self.config.expiration_time,
last_refreshed: SystemTime::now(),
metadata: request.metadata.clone(),
priority: request.priority,
wait_start_time: None,
},
Duration::ZERO,
))
} else {
// Update statistics
self.update_stats(false).await;
Err(LockError::QuorumNotReached {
required: self.config.quorum,
achieved: successful_responses,
})
}
}
/// Release distributed lock
pub async fn release_lock(&self, lock_id: &LockId, owner: &str) -> Result<LockResponse> {
let resource_key = self.get_resource_key_from_lock_id(lock_id);
// Check if we hold this lock
{
let locks = self.locks.read().await;
if let Some(lock) = locks.get(&resource_key) {
if lock.owner != owner {
return Err(LockError::NotOwner {
lock_id: lock_id.clone(),
owner: owner.to_string(),
});
}
}
}
// Release lock from all nodes
let clients = self.clients.read().await;
let mut responses = Vec::new();
for (node_id, client) in clients.iter() {
match client.release(lock_id).await {
Ok(response) => {
responses.push((node_id.clone(), response));
}
Err(e) => {
tracing::warn!("Lock release failed for node {}: {}", node_id, e);
}
}
}
// Remove lock entry
{
let mut locks = self.locks.write().await;
if let Some(lock) = locks.remove(&resource_key) {
// Cancel refresh task
if let Some(handle) = lock.refresh_handle {
handle.abort();
}
}
}
Ok(LockResponse::success(
LockInfo {
id: lock_id.clone(),
resource: "unknown".to_string(),
lock_type: LockType::Exclusive,
status: LockStatus::Released,
owner: owner.to_string(),
acquired_at: SystemTime::now(),
expires_at: SystemTime::now(),
last_refreshed: SystemTime::now(),
metadata: crate::types::LockMetadata::default(),
priority: LockPriority::Normal,
wait_start_time: None,
},
Duration::ZERO,
))
}
/// Start lock refresh task
async fn start_refresh_task(&self, lock_entry: &DistributedLockEntry) -> tokio::task::JoinHandle<()> {
let lock_id = lock_entry.lock_id.clone();
let _owner = lock_entry.owner.clone();
let _resource = lock_entry.resource.clone();
let refresh_interval = self.config.refresh_interval;
let clients = self.clients.clone();
let quorum = self.config.quorum;
tokio::spawn(async move {
let mut interval = interval(refresh_interval);
loop {
interval.tick().await;
// Try to refresh lock on all nodes
let clients_guard = clients.read().await;
let mut success_count = 0;
for (node_id, client) in clients_guard.iter() {
match client.refresh(&lock_id).await {
Ok(success) if success => {
success_count += 1;
}
_ => {
tracing::warn!("Lock refresh failed for node {}", node_id);
}
}
}
// If we don't have quorum, stop refreshing
if success_count < quorum {
tracing::error!("Lost quorum for lock {}, stopping refresh", lock_id);
break;
}
}
})
}
/// Get resource key
fn get_resource_key(&self, resource: &str) -> String {
format!("{}:{}", self.node_id, resource)
}
/// Get resource key from lock ID
fn get_resource_key_from_lock_id(&self, lock_id: &LockId) -> String {
// This is a simplified implementation
// In practice, you might want to store a mapping from lock_id to resource
format!("{}:{}", self.node_id, lock_id)
}
/// Update statistics
async fn update_stats(&self, success: bool) {
let mut stats = self.stats.lock().await;
if success {
stats.successful_acquires += 1;
} else {
stats.failed_acquires += 1;
}
}
/// Get lock statistics
pub async fn get_stats(&self) -> LockStats {
self.stats.lock().await.clone()
}
/// Clean up expired locks
pub async fn cleanup_expired_locks(&self) -> usize {
let mut locks = self.locks.write().await;
let initial_len = locks.len();
locks.retain(|_, lock| !lock.has_expired());
initial_len - locks.len()
}
/// Force release lock (admin operation)
pub async fn force_release_lock(&self, lock_id: &LockId) -> Result<LockResponse> {
let resource_key = self.get_resource_key_from_lock_id(lock_id);
// Check if we hold this lock
{
let locks = self.locks.read().await;
if let Some(lock) = locks.get(&resource_key) {
// Force release on all nodes
let clients = self.clients.read().await;
let mut _success_count = 0;
for (node_id, client) in clients.iter() {
match client.force_release(lock_id).await {
Ok(success) if success => {
_success_count += 1;
}
_ => {
tracing::warn!("Force release failed for node {}", node_id);
}
}
}
// Remove from local locks
let mut locks = self.locks.write().await;
locks.remove(&resource_key);
// Wake up waiting locks
return Ok(LockResponse::success(
LockInfo {
id: lock_id.clone(),
resource: lock.resource.clone(),
lock_type: lock.lock_type,
status: LockStatus::ForceReleased,
owner: lock.owner.clone(),
acquired_at: SystemTime::now(),
expires_at: SystemTime::now(),
last_refreshed: SystemTime::now(),
metadata: crate::types::LockMetadata::default(),
priority: lock.priority,
wait_start_time: None,
},
Duration::ZERO,
));
}
}
Err(LockError::internal("Lock not found"))
}
/// Refresh lock
pub async fn refresh_lock(&self, lock_id: &LockId, owner: &str) -> Result<LockResponse> {
let resource_key = self.get_resource_key_from_lock_id(lock_id);
// Check if we hold this lock
{
let locks = self.locks.read().await;
if let Some(lock) = locks.get(&resource_key) {
if lock.owner == owner && !lock.has_expired() {
// 提前 clone 所需字段
let priority = lock.priority;
let lock_id = lock.lock_id.clone();
let resource = lock.resource.clone();
let lock_type = lock.lock_type;
let owner = lock.owner.clone();
let holders = lock.holders.clone();
let acquired_at = lock.acquired_at;
let expires_at = Instant::now() + self.config.expiration_time;
// 更新锁
let mut locks = self.locks.write().await;
locks.insert(
resource_key.clone(),
DistributedLockEntry {
lock_id: lock_id.clone(),
resource: resource.clone(),
lock_type,
owner: owner.clone(),
priority,
acquired_at,
expires_at,
holders,
refresh_handle: None,
},
);
return Ok(LockResponse::success(
LockInfo {
id: lock_id,
resource,
lock_type,
status: LockStatus::Acquired,
owner,
acquired_at: SystemTime::now(),
expires_at: SystemTime::now() + self.config.expiration_time,
last_refreshed: SystemTime::now(),
metadata: crate::types::LockMetadata::default(),
priority,
wait_start_time: None,
},
Duration::ZERO,
));
}
}
}
Err(LockError::internal("Lock not found or expired"))
}
}
impl Default for DistributedLockManager {
fn default() -> Self {
Self::new(QuorumConfig::new(3, 1), Uuid::new_v4().to_string())
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::types::LockType;
#[tokio::test]
async fn test_quorum_config() {
let config = QuorumConfig::new(5, 2);
assert!(config.is_valid());
assert_eq!(config.quorum, 3);
}
#[tokio::test]
async fn test_distributed_lock_entry() {
let lock_id = LockId::new("test-resource");
let entry = DistributedLockEntry::new(
lock_id.clone(),
"test-resource".to_string(),
LockType::Exclusive,
"test-owner".to_string(),
LockPriority::Normal,
Duration::from_secs(60),
);
assert_eq!(entry.lock_id, lock_id);
assert!(!entry.has_expired());
assert!(entry.remaining_time() > Duration::ZERO);
}
#[tokio::test]
async fn test_distributed_lock_manager_creation() {
let config = QuorumConfig::new(3, 1);
let manager = DistributedLockManager::new(config, "node-1".to_string());
let stats = manager.get_stats().await;
assert_eq!(stats.successful_acquires, 0);
assert_eq!(stats.failed_acquires, 0);
}
#[tokio::test]
async fn test_force_release_lock() {
let config = QuorumConfig::new(3, 1);
let manager = DistributedLockManager::new(config, "node-1".to_string());
let lock_id = LockId::new("test-resource");
let result = manager.force_release_lock(&lock_id).await;
// Should fail because lock doesn't exist
assert!(result.is_err());
}
}
crates/lock/src/error.rs
+55
View File
@@ -88,6 +88,61 @@ pub enum LockError {
NotOwner { lock_id: LockId, owner: String },
}
impl Clone for LockError {
fn clone(&self) -> Self {
match self {
LockError::Timeout { resource, timeout } => LockError::Timeout {
resource: resource.clone(),
timeout: *timeout,
},
LockError::ResourceNotFound { resource } => LockError::ResourceNotFound {
resource: resource.clone(),
},
LockError::PermissionDenied { reason } => LockError::PermissionDenied { reason: reason.clone() },
LockError::Network { message, source: _ } => LockError::Network {
message: message.clone(),
source: Box::new(std::io::Error::other(message.clone())),
},
LockError::Internal { message } => LockError::Internal {
message: message.clone(),
},
LockError::AlreadyLocked { resource, owner } => LockError::AlreadyLocked {
resource: resource.clone(),
owner: owner.clone(),
},
LockError::InvalidHandle { handle_id } => LockError::InvalidHandle {
handle_id: handle_id.clone(),
},
LockError::Configuration { message } => LockError::Configuration {
message: message.clone(),
},
LockError::Serialization { message, source: _ } => LockError::Serialization {
message: message.clone(),
source: Box::new(std::io::Error::other(message.clone())),
},
LockError::Deserialization { message, source: _ } => LockError::Deserialization {
message: message.clone(),
source: Box::new(std::io::Error::other(message.clone())),
},
LockError::InsufficientNodes { required, available } => LockError::InsufficientNodes {
required: *required,
available: *available,
},
LockError::QuorumNotReached { required, achieved } => LockError::QuorumNotReached {
required: *required,
achieved: *achieved,
},
LockError::QueueFull { message } => LockError::QueueFull {
message: message.clone(),
},
LockError::NotOwner { lock_id, owner } => LockError::NotOwner {
lock_id: lock_id.clone(),
owner: owner.clone(),
},
}
}
}
impl LockError {
/// Create timeout error
pub fn timeout(resource: impl Into<String>, timeout: Duration) -> Self {
crates/lock/src/lib.rs
+7 -63
View File
@@ -1,4 +1,4 @@
#![allow(dead_code)]
// #![allow(dead_code)]
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
@@ -13,8 +13,6 @@
// See the License for the specific language governing permissions and
// limitations under the License.
// ============================================================================
// Core Module Declarations
// ============================================================================
@@ -25,14 +23,10 @@ pub mod namespace;
// Abstraction Layer Modules
pub mod client;
// Distributed Layer Modules
pub mod distributed;
// Local Layer Modules
pub mod local;
// Core Modules
pub mod config;
pub mod error;
pub mod types;
@@ -43,15 +37,12 @@ pub mod types;
// Re-export main types for easy access
pub use crate::{
// Client interfaces
client::{LockClient, local::LocalClient, remote::{RemoteClient, LockArgs}},
// Configuration
config::{DistributedLockConfig, LocalLockConfig, LockConfig, NetworkConfig},
distributed::{DistributedLockEntry, DistributedLockManager, QuorumConfig},
client::{LockClient, local::LocalClient, remote::RemoteClient},
// Error types
error::{LockError, Result},
local::LocalLockMap,
// Main components
namespace::{NamespaceLock, NamespaceLockManager, NsLockMap},
namespace::{NamespaceLock, NamespaceLockManager},
// Core types
types::{
HealthInfo, HealthStatus, LockId, LockInfo, LockMetadata, LockPriority, LockRequest, LockResponse, LockStats, LockStatus,
@@ -87,60 +78,13 @@ pub fn get_global_lock_map() -> Arc<local::LocalLockMap> {
GLOBAL_LOCK_MAP.get_or_init(|| Arc::new(local::LocalLockMap::new())).clone()
}
// ============================================================================
// Feature Flags
// ============================================================================
#[cfg(feature = "distributed")]
pub mod distributed_features {
// Distributed locking specific features
}
#[cfg(feature = "metrics")]
pub mod metrics {
// Metrics collection features
}
#[cfg(feature = "tracing")]
pub mod tracing_features {
// Tracing features
}
// ============================================================================
// Convenience Functions
// ============================================================================
/// Create a new namespace lock
pub fn create_namespace_lock(namespace: String, distributed: bool) -> NamespaceLock {
if distributed {
// Create a namespace lock that uses RPC to communicate with the server
// This will use the NsLockMap with distributed mode enabled
NamespaceLock::new(namespace, true)
} else {
NamespaceLock::new(namespace, false)
}
}
// ============================================================================
// Utility Functions
// ============================================================================
/// Generate a new lock ID
pub fn generate_lock_id() -> LockId {
LockId::new_deterministic("default")
}
/// Create a lock request with default settings
pub fn create_lock_request(resource: String, lock_type: LockType, owner: String) -> LockRequest {
LockRequest::new(resource, lock_type, owner)
}
/// Create an exclusive lock request
pub fn create_exclusive_lock_request(resource: String, owner: String) -> LockRequest {
create_lock_request(resource, LockType::Exclusive, owner)
}
/// Create a shared lock request
pub fn create_shared_lock_request(resource: String, owner: String) -> LockRequest {
create_lock_request(resource, LockType::Shared, owner)
pub fn create_namespace_lock(namespace: String, _distributed: bool) -> NamespaceLock {
// The distributed behavior is now determined by the type of clients added to the NamespaceLock
// This function just creates an empty NamespaceLock
NamespaceLock::new(namespace)
}
crates/lock/src/local.rs
+724 -528
View File
File diff suppressed because it is too large Load Diff
crates/lock/src/namespace.rs
+178 -754
View File
File diff suppressed because it is too large Load Diff
crates/lock/src/types.rs
+17 -17
View File
@@ -242,14 +242,14 @@ pub struct LockRequest {
pub lock_type: LockType,
/// Lock owner
pub owner: String,
/// Timeout duration
pub timeout: Duration,
/// Acquire timeout duration (how long to wait for lock acquisition)
pub acquire_timeout: Duration,
/// Lock TTL (Time To Live - how long the lock remains valid after acquisition)
pub ttl: Duration,
/// Lock metadata
pub metadata: LockMetadata,
/// Lock priority
pub priority: LockPriority,
/// Wait timeout duration
pub wait_timeout: Option<Duration>,
/// Deadlock detection
pub deadlock_detection: bool,
}
@@ -263,17 +263,23 @@ impl LockRequest {
resource: resource_str,
lock_type,
owner: owner.into(),
timeout: Duration::from_secs(30),
acquire_timeout: Duration::from_secs(10), // Default 10 seconds to acquire
ttl: Duration::from_secs(30), // Default 30 seconds lock lifetime
metadata: LockMetadata::default(),
priority: LockPriority::default(),
wait_timeout: None,
deadlock_detection: false,
}
}
/// Set timeout
pub fn with_timeout(mut self, timeout: Duration) -> Self {
self.timeout = timeout;
/// Set acquire timeout (how long to wait for lock acquisition)
pub fn with_acquire_timeout(mut self, timeout: Duration) -> Self {
self.acquire_timeout = timeout;
self
}
/// Set lock TTL (how long the lock remains valid after acquisition)
pub fn with_ttl(mut self, ttl: Duration) -> Self {
self.ttl = ttl;
self
}
@@ -289,12 +295,6 @@ impl LockRequest {
self
}
/// Set wait timeout
pub fn with_wait_timeout(mut self, wait_timeout: Duration) -> Self {
self.wait_timeout = Some(wait_timeout);
self
}
/// Set deadlock detection
pub fn with_deadlock_detection(mut self, enabled: bool) -> Self {
self.deadlock_detection = enabled;
@@ -640,14 +640,14 @@ mod tests {
#[test]
fn test_lock_request() {
let request = LockRequest::new("test-resource", LockType::Exclusive, "test-owner")
.with_timeout(Duration::from_secs(60))
.with_acquire_timeout(Duration::from_secs(60))
.with_priority(LockPriority::High)
.with_deadlock_detection(true);
assert_eq!(request.resource, "test-resource");
assert_eq!(request.lock_type, LockType::Exclusive);
assert_eq!(request.owner, "test-owner");
assert_eq!(request.timeout, Duration::from_secs(60));
assert_eq!(request.acquire_timeout, Duration::from_secs(60));
assert_eq!(request.priority, LockPriority::High);
assert!(request.deadlock_detection);
}