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

fix: Increase lock acquire timeout for network storage reliability (#1548)

Browse Source
This commit is contained in:
houseme
2026-01-19 01:14:36 +08:00
committed by GitHub
parent c9e2d7da2a
commit a9f499282c
11 changed files with 96 additions and 57 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+1
View File
@@ -7985,6 +7985,7 @@ dependencies = [
"crossbeam-queue",
"futures",
"parking_lot",
"rustfs-utils",
"serde",
"serde_json",
"smallvec",
crates/lock/Cargo.toml
+1
View File
@@ -29,6 +29,7 @@ documentation = "https://docs.rs/rustfs-lock/latest/rustfs_lock/"
workspace = true
[dependencies]
rustfs-utils = { workspace = true }
async-trait.workspace = true
futures.workspace = true
serde.workspace = true
crates/lock/src/client/local.rs
+10 -13
View File
@@ -17,11 +17,8 @@ use std::sync::Arc;
use tokio::sync::RwLock;
use crate::{
GlobalLockManager,
client::LockClient,
error::Result,
fast_lock::{FastLockGuard, LockManager},
types::{LockId, LockInfo, LockMetadata, LockPriority, LockRequest, LockResponse, LockStats, LockType},
FastLockGuard, GlobalLockManager, LockClient, LockId, LockInfo, LockManager, LockMetadata, LockPriority, LockRequest,
LockResponse, LockStats, LockStatus, LockType, Result,
};
/// Local lock client using FastLock
@@ -54,12 +51,12 @@ impl Default for LocalClient {
impl LockClient for LocalClient {
async fn acquire_exclusive(&self, request: &LockRequest) -> Result<LockResponse> {
let lock_manager = self.get_lock_manager();
let lock_request = crate::fast_lock::ObjectLockRequest::new_write("", request.resource.clone(), request.owner.clone())
let lock_request = crate::ObjectLockRequest::new_write("", request.resource.clone(), request.owner.clone())
.with_acquire_timeout(request.acquire_timeout);
match lock_manager.acquire_lock(lock_request).await {
Ok(guard) => {
let lock_id = crate::types::LockId::new_deterministic(&request.resource);
let lock_id = LockId::new_deterministic(&request.resource);
// Store guard for later release
let mut guards = self.guard_storage.write().await;
@@ -98,12 +95,12 @@ impl LockClient for LocalClient {
async fn acquire_shared(&self, request: &LockRequest) -> Result<LockResponse> {
let lock_manager = self.get_lock_manager();
let lock_request = crate::fast_lock::ObjectLockRequest::new_read("", request.resource.clone(), request.owner.clone())
let lock_request = crate::ObjectLockRequest::new_read("", request.resource.clone(), request.owner.clone())
.with_acquire_timeout(request.acquire_timeout);
match lock_manager.acquire_lock(lock_request).await {
Ok(guard) => {
let lock_id = crate::types::LockId::new_deterministic(&request.resource);
let lock_id = LockId::new_deterministic(&request.resource);
// Store guard for later release
let mut guards = self.guard_storage.write().await;
@@ -166,14 +163,14 @@ impl LockClient for LocalClient {
if let Some(guard) = guards.get(lock_id) {
// We have an active guard for this lock
let lock_type = match guard.mode() {
crate::fast_lock::types::LockMode::Shared => crate::types::LockType::Shared,
crate::fast_lock::types::LockMode::Exclusive => crate::types::LockType::Exclusive,
crate::LockMode::Shared => LockType::Shared,
crate::LockMode::Exclusive => LockType::Exclusive,
};
Ok(Some(LockInfo {
id: lock_id.clone(),
resource: lock_id.resource.clone(),
lock_type,
status: crate::types::LockStatus::Acquired,
status: LockStatus::Acquired,
owner: guard.owner().to_string(),
acquired_at: std::time::SystemTime::now(),
expires_at: std::time::SystemTime::now() + std::time::Duration::from_secs(30),
@@ -207,7 +204,7 @@ impl LockClient for LocalClient {
#[cfg(test)]
mod tests {
use super::*;
use crate::types::LockType;
use crate::LockType;
#[tokio::test]
async fn test_local_client_acquire_exclusive() {
crates/lock/src/client/mod.rs
+5 -9
View File
@@ -15,14 +15,10 @@
pub mod local;
// pub mod remote;
use crate::{LockId, LockInfo, LockRequest, LockResponse, LockStats, LockType, Result};
use async_trait::async_trait;
use std::sync::Arc;
use crate::{
error::Result,
types::{LockId, LockInfo, LockRequest, LockResponse, LockStats},
};
/// Lock client trait
#[async_trait]
pub trait LockClient: Send + Sync + std::fmt::Debug {
@@ -35,8 +31,8 @@ pub trait LockClient: Send + Sync + std::fmt::Debug {
/// Acquire lock (generic method)
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,
LockType::Exclusive => self.acquire_exclusive(request).await,
LockType::Shared => self.acquire_shared(request).await,
}
}
@@ -83,13 +79,13 @@ impl ClientFactory {
#[cfg(test)]
mod tests {
use super::*;
use crate::types::LockType;
use crate::LockType;
#[tokio::test]
async fn test_local_client_basic_operations() {
let client = ClientFactory::create_local();
let request = crate::types::LockRequest::new("test-resource", LockType::Exclusive, "test-owner");
let request = LockRequest::new("test-resource", LockType::Exclusive, "test-owner");
// Test lock acquisition
let response = client.acquire_exclusive(&request).await;
crates/lock/src/error.rs
+1 -1
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::types::LockId;
use crate::LockId;
use std::time::Duration;
use thiserror::Error;
crates/lock/src/fast_lock/mod.rs
+12 -5
View File
@@ -42,19 +42,26 @@ pub use disabled_manager::DisabledLockManager;
pub use guard::FastLockGuard;
pub use manager::FastObjectLockManager;
pub use manager_trait::LockManager;
use std::time::Duration;
pub use types::*;
/// Default RustFS specific timeouts in seconds
pub(crate) const DEFAULT_RUSTFS_MAX_ACQUIRE_TIMEOUT: u64 = 120;
/// Default RustFS acquire timeout in seconds
pub(crate) const DEFAULT_RUSTFS_ACQUIRE_TIMEOUT: u64 = 60;
/// Default shard count (must be power of 2)
pub const DEFAULT_SHARD_COUNT: usize = 1024;
/// Default lock timeout
pub const DEFAULT_LOCK_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
pub const DEFAULT_LOCK_TIMEOUT: Duration = Duration::from_secs(30);
/// Default acquire timeout - increased for database workloads
pub const DEFAULT_ACQUIRE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
/// Default acquire timeout - increased for network block storage workloads (e.g., Hetzner Ceph)
pub const DEFAULT_ACQUIRE_TIMEOUT: Duration = Duration::from_secs(DEFAULT_RUSTFS_ACQUIRE_TIMEOUT);
/// Maximum acquire timeout for high-load scenarios
pub const MAX_ACQUIRE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(60);
pub const MAX_ACQUIRE_TIMEOUT: Duration = Duration::from_secs(DEFAULT_RUSTFS_MAX_ACQUIRE_TIMEOUT);
/// Lock cleanup interval
pub const CLEANUP_INTERVAL: std::time::Duration = std::time::Duration::from_secs(60);
pub const CLEANUP_INTERVAL: Duration = Duration::from_secs(60);
crates/lock/src/fast_lock/types.rs
+1 -2
View File
@@ -12,6 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::fast_lock::guard::FastLockGuard;
use serde::{Deserialize, Serialize};
use smartstring::SmartString;
use std::hash::{Hash, Hasher};
@@ -19,8 +20,6 @@ use std::sync::Arc;
use std::sync::OnceLock;
use std::time::{Duration, SystemTime};
use crate::fast_lock::guard::FastLockGuard;
/// Object key for version-aware locking
#[derive(Debug, Clone, PartialEq, Eq, Hash, PartialOrd, Ord)]
pub struct ObjectKey {
crates/lock/src/guard.rs
+1 -1
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::{client::LockClient, types::LockId};
use crate::{LockClient, LockId};
use std::sync::{Arc, LazyLock};
use tokio::sync::mpsc;
crates/lock/src/lib.rs
+37 -13
View File
@@ -68,11 +68,15 @@ pub const BUILD_TIMESTAMP: &str = "unknown";
/// Maximum number of items in delete list
pub const MAX_DELETE_LIST: usize = 1000;
/// Default setting for RUSTFS_ENABLE_LOCKS environment variable
const DEFAULT_RUSTFS_LOCKS_ENABLED: bool = true;
// ============================================================================
// Global FastLock Manager
// ============================================================================
// Global singleton FastLock manager shared across all lock implementations
use crate::fast_lock::{DEFAULT_RUSTFS_ACQUIRE_TIMEOUT, DEFAULT_RUSTFS_MAX_ACQUIRE_TIMEOUT};
use std::sync::Arc;
use std::sync::OnceLock;
@@ -92,20 +96,40 @@ impl GlobalLockManager {
/// Create a lock manager based on environment variable configuration
pub fn new() -> Self {
// Check RUSTFS_ENABLE_LOCKS environment variable
let locks_enabled = std::env::var("RUSTFS_ENABLE_LOCKS")
.unwrap_or_else(|_| "true".to_string())
.to_lowercase();
match locks_enabled.as_str() {
"false" | "0" | "no" | "off" | "disabled" => {
tracing::info!("Lock system disabled via RUSTFS_ENABLE_LOCKS environment variable");
Self::Disabled(DisabledLockManager::new())
}
_ => {
tracing::info!("Lock system enabled");
Self::Enabled(Arc::new(FastObjectLockManager::new()))
}
let locks_enabled = rustfs_utils::get_env_bool("RUSTFS_ENABLE_LOCKS", DEFAULT_RUSTFS_LOCKS_ENABLED);
if !locks_enabled {
tracing::info!("Lock system disabled via RUSTFS_ENABLE_LOCKS environment variable");
return Self::Disabled(DisabledLockManager::new());
}
tracing::info!("Lock system enabled");
// Read lock acquire timeout from environment variable
let mut acquire_secs = rustfs_utils::get_env_u64("RUSTFS_LOCK_ACQUIRE_TIMEOUT", DEFAULT_RUSTFS_ACQUIRE_TIMEOUT);
// Enforce minimum of 1 second
if acquire_secs == 0 {
tracing::warn!("Requested lock acquire timeout {}s is below minimum 1s, using minimum", acquire_secs);
acquire_secs = 1;
}
if acquire_secs > DEFAULT_RUSTFS_MAX_ACQUIRE_TIMEOUT {
tracing::warn!(
"Requested lock acquire timeout {}s exceeds maximum {}, using maximum",
acquire_secs,
DEFAULT_RUSTFS_MAX_ACQUIRE_TIMEOUT
);
acquire_secs = DEFAULT_RUSTFS_MAX_ACQUIRE_TIMEOUT;
}
let acquire_timeout = std::time::Duration::from_secs(acquire_secs);
tracing::info!("Lock system enabled with acquire timeout: {}s", acquire_timeout.as_secs());
// Create lock manager with custom configuration
let config = fast_lock::LockConfig {
default_acquire_timeout: acquire_timeout,
..Default::default()
};
Self::Enabled(Arc::new(FastObjectLockManager::with_config(config)))
}
/// Check if the lock manager is disabled
crates/utils/src/envs.rs
+18 -13
View File
@@ -363,14 +363,23 @@ pub fn get_env_opt_str(key: &str) -> Option<String> {
/// - `bool`: The parsed boolean value if successful, otherwise the default value.
///
pub fn get_env_bool(key: &str, default: bool) -> bool {
env::var(key)
.ok()
.and_then(|v| match v.to_lowercase().as_str() {
"1" | "t" | "T" | "true" | "TRUE" | "True" | "on" | "ON" | "On" | "enabled" => Some(true),
"0" | "f" | "F" | "false" | "FALSE" | "False" | "off" | "OFF" | "Off" | "disabled" => Some(false),
_ => None,
})
.unwrap_or(default)
env::var(key).ok().and_then(|v| parse_bool_str(&v)).unwrap_or(default)
}
/// Parse a string into a boolean value.
///
/// #Parameters
/// - `s`: The string to parse.
///
/// #Returns
/// - `Option<bool>`: The parsed boolean value if successful, otherwise None.
///
fn parse_bool_str(s: &str) -> Option<bool> {
match s.trim().to_ascii_lowercase().as_str() {
"1" | "t" | "true" | "on" | "yes" | "ok" | "success" | "active" | "enabled" => Some(true),
"0" | "f" | "false" | "off" | "no" | "not_ok" | "failure" | "inactive" | "disabled" => Some(false),
_ => None,
}
}
/// Retrieve an environment variable as a boolean, returning None if not set or parsing fails.
@@ -382,9 +391,5 @@ pub fn get_env_bool(key: &str, default: bool) -> bool {
/// - `Option<bool>`: The parsed boolean value if successful, otherwise None.
///
pub fn get_env_opt_bool(key: &str) -> Option<bool> {
env::var(key).ok().and_then(|v| match v.to_lowercase().as_str() {
"1" | "t" | "T" | "true" | "TRUE" | "True" | "on" | "ON" | "On" | "enabled" => Some(true),
"0" | "f" | "F" | "false" | "FALSE" | "False" | "off" | "OFF" | "Off" | "disabled" => Some(false),
_ => None,
})
env::var(key).ok().and_then(|v| parse_bool_str(&v))
}
scripts/run.sh
+9
View File
@@ -189,6 +189,15 @@ export RUSTFS_TRUST_SYSTEM_CA=true
# Enable FTP server
export RUSTFS_FTPS_ENABLE=false
# Use default timeout (60 seconds)
# No environment variable needed
# Increase timeout for high-latency network storage
#export RUSTFS_LOCK_ACQUIRE_TIMEOUT=120
# Reduce timeout for low-latency local storage
export RUSTFS_LOCK_ACQUIRE_TIMEOUT=30
if [ -n "$1" ]; then
export RUSTFS_VOLUMES="$1"
fi