mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-21 20:06:37 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d33e2fdcf1 |
@@ -57,13 +57,6 @@ pub const DEFAULT_MAX_IO_EVENTS_PER_TICK: usize = 1024;
|
|||||||
pub const DEFAULT_EVENT_INTERVAL: u32 = 61;
|
pub const DEFAULT_EVENT_INTERVAL: u32 = 61;
|
||||||
pub const DEFAULT_RNG_SEED: Option<u64> = None; // None means random
|
pub const DEFAULT_RNG_SEED: Option<u64> = None; // None means random
|
||||||
|
|
||||||
/// Dedicated blocking thread pool for fsync/fdatasync operations.
|
|
||||||
/// When > 1, fsync operations are isolated from the main blocking pool to
|
|
||||||
/// prevent device-bound fsync from starving read operations (pread/stat/open).
|
|
||||||
/// Default 0 means auto (no isolation, use main runtime).
|
|
||||||
pub const ENV_FSYNC_BLOCKING_THREADS: &str = "RUSTFS_RUNTIME_FSYNC_BLOCKING_THREADS";
|
|
||||||
pub const DEFAULT_FSYNC_BLOCKING_THREADS: usize = 0;
|
|
||||||
|
|
||||||
// Dial9 Tokio Telemetry Default values
|
// Dial9 Tokio Telemetry Default values
|
||||||
pub const DEFAULT_RUNTIME_DIAL9_ENABLED: bool = false; // Disabled by default
|
pub const DEFAULT_RUNTIME_DIAL9_ENABLED: bool = false; // Disabled by default
|
||||||
pub const DEFAULT_RUNTIME_DIAL9_OUTPUT_DIR: &str = "/var/log/rustfs/telemetry";
|
pub const DEFAULT_RUNTIME_DIAL9_OUTPUT_DIR: &str = "/var/log/rustfs/telemetry";
|
||||||
|
|||||||
@@ -315,7 +315,7 @@ pub async fn fsync_dir(dir: impl AsRef<Path>) -> io::Result<()> {
|
|||||||
#[cfg(unix)]
|
#[cfg(unix)]
|
||||||
{
|
{
|
||||||
let dir = dir.as_ref().to_path_buf();
|
let dir = dir.as_ref().to_path_buf();
|
||||||
fsync_spawn_blocking(move || fsync_dir_std(dir)).await?
|
tokio::task::spawn_blocking(move || fsync_dir_std(dir)).await?
|
||||||
}
|
}
|
||||||
|
|
||||||
#[cfg(not(unix))]
|
#[cfg(not(unix))]
|
||||||
@@ -683,7 +683,7 @@ async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
let dir = group.dir.clone();
|
let dir = group.dir.clone();
|
||||||
let dir_file = group.dir_file.clone();
|
let dir_file = group.dir_file.clone();
|
||||||
fsync_spawn_blocking(move || {
|
tokio::task::spawn_blocking(move || {
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
{
|
{
|
||||||
if let Some(kind) = fsync_dir_recorder::take_grouped_failure(&dir) {
|
if let Some(kind) = fsync_dir_recorder::take_grouped_failure(&dir) {
|
||||||
@@ -1080,44 +1080,6 @@ const TEST_GLOBAL_FILE_SYNCS: usize = 64;
|
|||||||
|
|
||||||
static FILE_SYNC_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
|
static FILE_SYNC_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
|
||||||
static DISK_FILE_SYNC_LIMITERS: LazyLock<Mutex<HashMap<PathBuf, Weak<Semaphore>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
|
static DISK_FILE_SYNC_LIMITERS: LazyLock<Mutex<HashMap<PathBuf, Weak<Semaphore>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||||
|
|
||||||
/// Dedicated tokio runtime for fsync/fdatasync blocking operations. When
|
|
||||||
/// configured with >1 threads, isolates device-bound fsync from the main
|
|
||||||
/// blocking pool so reads (pread/stat/open) are not starved. `None` means
|
|
||||||
/// fall back to the main runtime (zero behavior change).
|
|
||||||
static FSYNC_RUNTIME: LazyLock<Option<tokio::runtime::Runtime>> = LazyLock::new(|| {
|
|
||||||
let threads =
|
|
||||||
rustfs_utils::get_env_usize(rustfs_config::ENV_FSYNC_BLOCKING_THREADS, rustfs_config::DEFAULT_FSYNC_BLOCKING_THREADS);
|
|
||||||
if threads <= 1 {
|
|
||||||
return None;
|
|
||||||
}
|
|
||||||
let mut builder = tokio::runtime::Builder::new_multi_thread();
|
|
||||||
builder
|
|
||||||
.worker_threads(2)
|
|
||||||
.max_blocking_threads(threads)
|
|
||||||
.thread_name("rustfs-fsync")
|
|
||||||
.thread_stack_size(512 * 1024)
|
|
||||||
.enable_all();
|
|
||||||
match builder.build() {
|
|
||||||
Ok(rt) => {
|
|
||||||
tracing::info!(threads, "fsync dedicated blocking pool enabled");
|
|
||||||
Some(rt)
|
|
||||||
}
|
|
||||||
Err(err) => {
|
|
||||||
tracing::warn!(%err, "failed to build fsync runtime, falling back to main pool");
|
|
||||||
None
|
|
||||||
}
|
|
||||||
}
|
|
||||||
});
|
|
||||||
|
|
||||||
/// Spawn a blocking task on the fsync-dedicated runtime if configured,
|
|
||||||
/// otherwise fall back to the main tokio blocking pool.
|
|
||||||
fn fsync_spawn_blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> tokio::task::JoinHandle<T> {
|
|
||||||
match FSYNC_RUNTIME.as_ref() {
|
|
||||||
Some(rt) => rt.spawn_blocking(f),
|
|
||||||
None => tokio::task::spawn_blocking(f),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
|
static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
|
||||||
LazyLock::new(|| Mutex::new(HashMap::new()));
|
LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||||
type NamespaceMutationLock = AsyncMutex<()>;
|
type NamespaceMutationLock = AsyncMutex<()>;
|
||||||
@@ -1255,7 +1217,7 @@ where
|
|||||||
F: FnOnce() -> io::Result<T> + Send + 'static,
|
F: FnOnce() -> io::Result<T> + Send + 'static,
|
||||||
{
|
{
|
||||||
let (disk_permit, global_permit) = acquire_file_sync_permits(disk_permits).await?;
|
let (disk_permit, global_permit) = acquire_file_sync_permits(disk_permits).await?;
|
||||||
let result = fsync_spawn_blocking(move || {
|
let result = tokio::task::spawn_blocking(move || {
|
||||||
let _disk_permit = disk_permit;
|
let _disk_permit = disk_permit;
|
||||||
work()
|
work()
|
||||||
})
|
})
|
||||||
@@ -2184,7 +2146,7 @@ async fn run_blocking_namespace_file_sync_operation_with_global<T: Send + 'stati
|
|||||||
wait_started,
|
wait_started,
|
||||||
);
|
);
|
||||||
let disk_permit = admission.disk_permit.clone();
|
let disk_permit = admission.disk_permit.clone();
|
||||||
let result = fsync_spawn_blocking(move || {
|
let result = tokio::task::spawn_blocking(move || {
|
||||||
let _lease = lease;
|
let _lease = lease;
|
||||||
let _disk_permit = disk_permit;
|
let _disk_permit = disk_permit;
|
||||||
operation()
|
operation()
|
||||||
|
|||||||
+32
-138
@@ -12,18 +12,16 @@
|
|||||||
// See the License for the specific language governing permissions and
|
// See the License for the specific language governing permissions and
|
||||||
// limitations under the License.
|
// limitations under the License.
|
||||||
|
|
||||||
use std::{collections::HashMap, time::Duration};
|
|
||||||
|
|
||||||
use jiff::Timestamp;
|
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
use crate::heal_commands::HealResultItem;
|
/// Bitflag helper for service trace categories.
|
||||||
|
///
|
||||||
|
/// Each variant occupies a single bit so that a `TraceType` value can represent
|
||||||
|
/// an arbitrary combination of categories via bitwise OR.
|
||||||
#[derive(Debug, Clone, Copy, Serialize, Deserialize, Default)]
|
#[derive(Debug, Clone, Copy, Serialize, Deserialize, Default)]
|
||||||
pub struct TraceType(u64);
|
pub struct TraceType(u64);
|
||||||
|
|
||||||
impl TraceType {
|
impl TraceType {
|
||||||
// Define some constants
|
|
||||||
pub const OS: TraceType = TraceType(1 << 0);
|
pub const OS: TraceType = TraceType(1 << 0);
|
||||||
pub const STORAGE: TraceType = TraceType(1 << 1);
|
pub const STORAGE: TraceType = TraceType(1 << 1);
|
||||||
pub const S3: TraceType = TraceType(1 << 2);
|
pub const S3: TraceType = TraceType(1 << 2);
|
||||||
@@ -40,15 +38,13 @@ impl TraceType {
|
|||||||
pub const FTP: TraceType = TraceType(1 << 13);
|
pub const FTP: TraceType = TraceType(1 << 13);
|
||||||
pub const ILM: TraceType = TraceType(1 << 14);
|
pub const ILM: TraceType = TraceType(1 << 14);
|
||||||
|
|
||||||
// MetricsAll must be last.
|
/// All trace categories combined. Must be updated when adding new variants.
|
||||||
pub const ALL: TraceType = TraceType((1 << 15) - 1);
|
pub const ALL: TraceType = TraceType((1 << 15) - 1);
|
||||||
|
|
||||||
pub fn new(t: u64) -> Self {
|
pub fn new(t: u64) -> Self {
|
||||||
Self(t)
|
Self(t)
|
||||||
}
|
}
|
||||||
}
|
|
||||||
|
|
||||||
impl TraceType {
|
|
||||||
pub fn contains(&self, x: &TraceType) -> bool {
|
pub fn contains(&self, x: &TraceType) -> bool {
|
||||||
(self.0 & x.0) == x.0
|
(self.0 & x.0) == x.0
|
||||||
}
|
}
|
||||||
@@ -76,140 +72,38 @@ impl TraceType {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
|
||||||
pub struct TraceInfo {
|
|
||||||
#[serde(rename = "type")]
|
|
||||||
trace_type: u64,
|
|
||||||
#[serde(rename = "nodename")]
|
|
||||||
node_name: String,
|
|
||||||
#[serde(rename = "funcname")]
|
|
||||||
func_name: String,
|
|
||||||
#[serde(rename = "time")]
|
|
||||||
time: Timestamp,
|
|
||||||
#[serde(rename = "path")]
|
|
||||||
path: String,
|
|
||||||
#[serde(rename = "dur")]
|
|
||||||
duration: Duration,
|
|
||||||
#[serde(rename = "bytes", skip_serializing_if = "Option::is_none")]
|
|
||||||
bytes: Option<i64>,
|
|
||||||
#[serde(rename = "msg", skip_serializing_if = "Option::is_none")]
|
|
||||||
message: Option<String>,
|
|
||||||
#[serde(rename = "error", skip_serializing_if = "Option::is_none")]
|
|
||||||
error: Option<String>,
|
|
||||||
#[serde(rename = "custom", skip_serializing_if = "Option::is_none")]
|
|
||||||
custom: Option<HashMap<String, String>>,
|
|
||||||
#[serde(rename = "http", skip_serializing_if = "Option::is_none")]
|
|
||||||
http: Option<TraceHTTPStats>,
|
|
||||||
#[serde(rename = "healResult", skip_serializing_if = "Option::is_none")]
|
|
||||||
heal_result: Option<HealResultItem>,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl TraceInfo {
|
|
||||||
pub fn mask(&self) -> u64 {
|
|
||||||
TraceType::new(self.trace_type).mask()
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
|
||||||
pub struct TraceInfoLegacy {
|
|
||||||
trace_info: TraceInfo,
|
|
||||||
#[serde(rename = "request")]
|
|
||||||
req_info: Option<TraceRequestInfo>,
|
|
||||||
#[serde(rename = "response")]
|
|
||||||
resp_info: Option<TraceResponseInfo>,
|
|
||||||
#[serde(rename = "stats")]
|
|
||||||
call_stats: Option<TraceCallStats>,
|
|
||||||
#[serde(rename = "storageStats")]
|
|
||||||
storage_stats: Option<StorageStats>,
|
|
||||||
#[serde(rename = "osStats")]
|
|
||||||
os_stats: Option<OSStats>,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
|
||||||
pub struct StorageStats {
|
|
||||||
path: String,
|
|
||||||
duration: Duration,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
|
||||||
pub struct OSStats {
|
|
||||||
path: String,
|
|
||||||
duration: Duration,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
|
||||||
pub struct TraceHTTPStats {
|
|
||||||
req_info: TraceRequestInfo,
|
|
||||||
resp_info: TraceResponseInfo,
|
|
||||||
call_stats: TraceCallStats,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
|
||||||
pub struct TraceCallStats {
|
|
||||||
input_bytes: i32,
|
|
||||||
output_bytes: i32,
|
|
||||||
latency: Duration,
|
|
||||||
time_to_first_byte: Duration,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
|
||||||
pub struct TraceRequestInfo {
|
|
||||||
time: Timestamp,
|
|
||||||
proto: String,
|
|
||||||
method: String,
|
|
||||||
#[serde(skip_serializing_if = "Option::is_none")]
|
|
||||||
path: Option<String>,
|
|
||||||
#[serde(skip_serializing_if = "Option::is_none")]
|
|
||||||
raw_query: Option<String>,
|
|
||||||
#[serde(skip_serializing_if = "Option::is_none")]
|
|
||||||
headers: Option<HashMap<String, String>>,
|
|
||||||
#[serde(skip_serializing_if = "Option::is_none")]
|
|
||||||
body: Option<Vec<u8>>,
|
|
||||||
client: String,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
|
|
||||||
pub struct TraceResponseInfo {
|
|
||||||
time: Timestamp,
|
|
||||||
#[serde(skip_serializing_if = "Option::is_none")]
|
|
||||||
headers: Option<HashMap<String, String>>,
|
|
||||||
#[serde(skip_serializing_if = "Option::is_none")]
|
|
||||||
body: Option<Vec<u8>>,
|
|
||||||
#[serde(skip_serializing_if = "Option::is_none")]
|
|
||||||
status_code: Option<i32>,
|
|
||||||
}
|
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn trace_timestamps_serialize_as_rfc3339_utc() {
|
fn trace_type_contains_and_overlaps() {
|
||||||
let timestamp = Timestamp::constant(1_700_000_000, 123_456_000);
|
let mut combined = TraceType::default();
|
||||||
let trace = TraceInfo {
|
combined.merge(&TraceType::S3);
|
||||||
time: timestamp,
|
combined.merge(&TraceType::HEALING);
|
||||||
http: Some(TraceHTTPStats {
|
|
||||||
req_info: TraceRequestInfo {
|
|
||||||
time: timestamp,
|
|
||||||
..Default::default()
|
|
||||||
},
|
|
||||||
resp_info: TraceResponseInfo {
|
|
||||||
time: timestamp,
|
|
||||||
..Default::default()
|
|
||||||
},
|
|
||||||
..Default::default()
|
|
||||||
}),
|
|
||||||
..Default::default()
|
|
||||||
};
|
|
||||||
|
|
||||||
let value = serde_json::to_value(trace).expect("trace should serialize");
|
assert!(combined.contains(&TraceType::S3));
|
||||||
assert_eq!(value["time"], "2023-11-14T22:13:20.123456Z");
|
assert!(combined.contains(&TraceType::HEALING));
|
||||||
assert_eq!(value["http"]["req_info"]["time"], "2023-11-14T22:13:20.123456Z");
|
assert!(!combined.contains(&TraceType::SCANNER));
|
||||||
assert_eq!(value["http"]["resp_info"]["time"], "2023-11-14T22:13:20.123456Z");
|
assert!(combined.overlaps(&TraceType::S3));
|
||||||
let trace: TraceInfo = serde_json::from_value(value).expect("trace should deserialize");
|
assert!(combined.overlaps(&TraceType::HEALING));
|
||||||
assert_eq!(trace.time, timestamp);
|
assert!(!combined.overlaps(&TraceType::SCANNER));
|
||||||
let http = trace.http.expect("http trace should deserialize");
|
}
|
||||||
assert_eq!(http.req_info.time, timestamp);
|
|
||||||
assert_eq!(http.resp_info.time, timestamp);
|
#[test]
|
||||||
|
fn trace_type_set_if() {
|
||||||
|
let mut tt = TraceType::default();
|
||||||
|
tt.set_if(true, &TraceType::OS);
|
||||||
|
tt.set_if(false, &TraceType::S3);
|
||||||
|
assert!(tt.contains(&TraceType::OS));
|
||||||
|
assert!(!tt.contains(&TraceType::S3));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn trace_type_single_type() {
|
||||||
|
assert!(TraceType::S3.single_type());
|
||||||
|
let mut combined = TraceType::S3;
|
||||||
|
combined.merge(&TraceType::HEALING);
|
||||||
|
assert!(!combined.single_type());
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user