diff --git a/Cargo.lock b/Cargo.lock index d00b4b1eac3..8260a8a7b39 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5545,6 +5545,7 @@ dependencies = [ "tokio", "tracing", "uuid", + "windows-sys 0.61.2", ] [[package]] diff --git a/crates/dbsp/Cargo.toml b/crates/dbsp/Cargo.toml index 8af54c76821..f0f5ec07ff0 100644 --- a/crates/dbsp/Cargo.toml +++ b/crates/dbsp/Cargo.toml @@ -99,10 +99,15 @@ feldera-buffer-cache = { workspace = true } memory-stats = { workspace = true } serde_json_path_to_error = { workspace = true } flate2 = { workspace = true } -nix = { workspace = true, features = ["process", "time"] } thread-id = { workspace = true } pin-project-lite = { workspace = true } +[target.'cfg(unix)'.dependencies] +nix = { workspace = true, features = ["process", "time"] } + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Storage_FileSystem", "Win32_System_IO", "Win32_System_Threading"] } + [dev-dependencies] rand = { workspace = true } rand_distr = { workspace = true } @@ -115,7 +120,6 @@ csv = { workspace = true } tar = { workspace = true } zstd = { workspace = true } criterion = { workspace = true } -pprof = { workspace = true, features = ["flamegraph", "criterion"] } rand_xoshiro = { workspace = true } indicatif = { workspace = true } reqwest = { workspace = true, features = ["blocking"] } @@ -123,6 +127,9 @@ chrono = { workspace = true, features = ["serde"] } tracing-subscriber = { version = "0.3.20", features = ["env-filter"] } tarpc = { workspace = true, features = ["full"] } +[target.'cfg(unix)'.dev-dependencies] +pprof = { workspace = true, features = ["flamegraph", "criterion"] } + [[bench]] name = "galen" harness = false diff --git a/crates/dbsp/benches/column_layer.rs b/crates/dbsp/benches/column_layer.rs index 6c41ee482b9..610249fee48 100644 --- a/crates/dbsp/benches/column_layer.rs +++ b/crates/dbsp/benches/column_layer.rs @@ -10,6 +10,7 @@ use dbsp::{ }, utils::consolidate, }; +#[cfg(unix)] use pprof::criterion::{Output, PProfProfiler}; use rand::{Rng, SeedableRng, distributions::Standard, prelude::Distribution}; use rand_xoshiro::Xoshiro256StarStar; @@ -233,9 +234,18 @@ leaf_benches! { "100,000,000-erased" = [Leaf]100_000_000, } +#[cfg(unix)] criterion_group!( name = benches; config = Criterion::default().with_profiler(PProfProfiler::new(300, Output::Flamegraph(None))); targets = column_leaf, merge_ordered_column_leaf_builder ); + +#[cfg(not(unix))] +criterion_group!( + name = benches; + config = Criterion::default(); + targets = column_leaf, merge_ordered_column_leaf_builder +); + criterion_main!(benches); diff --git a/crates/dbsp/src/circuit/circuit_builder.rs b/crates/dbsp/src/circuit/circuit_builder.rs index 8129d910204..577eef903f3 100644 --- a/crates/dbsp/src/circuit/circuit_builder.rs +++ b/crates/dbsp/src/circuit/circuit_builder.rs @@ -62,6 +62,7 @@ use feldera_ir::{LirCircuit, LirNodeId}; use feldera_samply::Span; use feldera_storage::{FileCommitter, StoragePath}; use itertools::Itertools; +#[cfg(unix)] use nix::{ sys::time::TimeValLike, time::{ClockId, clock_gettime}, @@ -93,9 +94,6 @@ use tracing::debug; use typedmap::{TypedMap, TypedMapKey}; use super::dbsp_handle::Mode; - -/// Label name used to store operator's persistent id, -/// i.e., id stable across circuit modifications. const LABEL_PERSISTENT_OPERATOR_ID: &str = "persistent_id"; /// Value stored in the stream. @@ -8783,24 +8781,71 @@ impl Timed { pub struct ThreadCpuTime(pub Duration); impl ThreadCpuTime { - /// Returns the current time elapsed running the current thread. + /// Returns the current thread's cumulative CPU time. pub fn now() -> Self { - let nanos = clock_gettime(ClockId::CLOCK_THREAD_CPUTIME_ID) - .unwrap() - .num_nanoseconds(); - Self(Duration::from_nanos(nanos.max(0).cast_unsigned())) + Self(current_thread_cpu_time()) } - /// Returns the time elapsed running the current thread since this - /// `ThreadCpuTime`. + /// Returns the CPU time elapsed on the current thread since this + /// `ThreadCpuTime` was recorded. /// /// This only makes sense if this `ThreadCpuTime` was for the currently /// running thread. /// /// Returns zero if the current time is earlier than self. pub fn elapsed(&self) -> Duration { - Self::now().0.saturating_sub(self.0) + current_thread_cpu_time().saturating_sub(self.0) + } +} + +#[cfg(unix)] +/// Returns the current thread's cumulative CPU time from `CLOCK_THREAD_CPUTIME_ID`. +fn current_thread_cpu_time() -> Duration { + let nanos = clock_gettime(ClockId::CLOCK_THREAD_CPUTIME_ID) + .unwrap() + .num_nanoseconds(); + Duration::from_nanos(nanos.max(0).cast_unsigned()) +} + +#[cfg(windows)] +/// Returns the current thread's cumulative kernel and user CPU time. +fn current_thread_cpu_time() -> Duration { + use windows_sys::Win32::{ + Foundation::FILETIME, + System::Threading::{GetCurrentThread, GetThreadTimes}, + }; + + let mut creation = FILETIME::default(); + let mut exit = FILETIME::default(); + let mut kernel = FILETIME::default(); + let mut user = FILETIME::default(); + // SAFETY: `GetCurrentThread` returns a valid pseudo-handle, and all four + // initialized FILETIME values remain valid mutable output buffers for the call. + let result = unsafe { + GetThreadTimes( + GetCurrentThread(), + &mut creation, + &mut exit, + &mut kernel, + &mut user, + ) + }; + if result == 0 { + // it is very unlikely this will fail, since GetCurrentThread() always returns a + // valid pseudo-handle with proper permissions + panic!("GetThreadTimes failed: {}", std::io::Error::last_os_error()); } + + let filetime_to_ticks = + |time: FILETIME| (u64::from(time.dwHighDateTime) << 32) | u64::from(time.dwLowDateTime); + let ticks = filetime_to_ticks(kernel).saturating_add(filetime_to_ticks(user)); + Duration::new(ticks / 10_000_000, ((ticks % 10_000_000) * 100) as u32) +} + +#[cfg(not(any(unix, windows)))] +/// Returns zero when the platform has no supported thread CPU clock. +fn current_thread_cpu_time() -> Duration { + Duration::ZERO } impl Future for Timed @@ -8831,6 +8876,15 @@ mod tests { use anyhow::anyhow; use std::{cell::RefCell, ops::Deref, rc::Rc, vec::Vec}; + #[test] + fn thread_cpu_time_is_monotonic() { + use crate::circuit::ThreadCpuTime; + + let start = ThreadCpuTime::now(); + std::hint::black_box((0..100_000u64).fold(0u64, |sum, value| sum.wrapping_add(value))); + assert!(ThreadCpuTime::now().0 >= start.0); + } + #[test] fn sum_circuit_dynamic() { sum_circuit::(); diff --git a/crates/dbsp/src/storage/backend.rs b/crates/dbsp/src/storage/backend.rs index 6e8e7f107c5..aa52caff88e 100644 --- a/crates/dbsp/src/storage/backend.rs +++ b/crates/dbsp/src/storage/backend.rs @@ -52,6 +52,8 @@ impl StorageCacheFlags for OpenOptions { use std::os::unix::fs::OpenOptionsExt; self.custom_flags(cache.to_custom_open_flags()); } + #[cfg(not(unix))] + let _ = cache; self } } diff --git a/crates/dbsp/src/storage/backend/posixio_impl.rs b/crates/dbsp/src/storage/backend/posixio_impl.rs index 38d3f6a8635..469a8b8d045 100644 --- a/crates/dbsp/src/storage/backend/posixio_impl.rs +++ b/crates/dbsp/src/storage/backend/posixio_impl.rs @@ -27,7 +27,6 @@ use std::time::{Duration, Instant}; use std::{ fs::{self, File, OpenOptions}, io::Error as IoError, - os::unix::fs::MetadataExt, path::{Path, PathBuf}, sync::{ Arc, @@ -36,10 +35,54 @@ use std::{ }; use tracing::{debug, warn}; +#[cfg(windows)] +/// Opens a separate overlapped reader handle for `file`. +/// +/// The returned handle is reused by every positioned read from this reader. +fn reader_file(file: &File) -> Result, IoError> { + use std::os::windows::io::{AsRawHandle, FromRawHandle}; + use windows_sys::Win32::{ + Foundation::INVALID_HANDLE_VALUE, + Storage::FileSystem::{ + FILE_FLAG_OVERLAPPED, FILE_GENERIC_READ, FILE_SHARE_DELETE, FILE_SHARE_READ, + FILE_SHARE_WRITE, ReOpenFile, + }, + }; + + // SAFETY: `file` owns a valid handle borrowed for the duration of this call. + let handle = unsafe { + ReOpenFile( + file.as_raw_handle(), + FILE_GENERIC_READ, + FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, + FILE_FLAG_OVERLAPPED, + ) + }; + if handle == INVALID_HANDLE_VALUE { + return Err(IoError::last_os_error()); + } + // SAFETY: ReOpenFile returned a distinct owned handle after excluding + // INVALID_HANDLE_VALUE. The returned File takes ownership exactly once. + Ok(Arc::new(unsafe { File::from_raw_handle(handle) })) +} + +#[cfg(windows)] +/// Reads exactly `len` bytes at `offset` through an overlapped `file` and +/// appends them to `buffer`. +fn read_exact_at_overlapped( + buffer: &mut FBuf, + file: &File, + offset: u64, + len: usize, +) -> Result<(), IoError> { + buffer.read_exact_at_overlapped(file, offset, len) +} + /// fsync the directory at `path` so a freshly-created child entry (a /// rename target or a new subdirectory) becomes durable. Without this, /// POSIX gives no guarantee that the directory entry survives a crash /// even if the child file itself has been fully fsynced. +#[cfg(unix)] fn fsync_dir(path: &Path) -> Result<(), StorageError> { let dir = File::open(path) .map_err(|e| StorageError::stdio(e.kind(), "open dir for fsync", path.display()))?; @@ -47,9 +90,69 @@ fn fsync_dir(path: &Path) -> Result<(), StorageError> { .map_err(|e| StorageError::stdio(e.kind(), "fsync dir", path.display())) } +#[cfg(windows)] +/// Flushes the directory at `path` so newly-created child entries are durable. +/// +/// Windows requires `FILE_FLAG_BACKUP_SEMANTICS` to open a directory handle. +fn fsync_dir(path: &Path) -> Result<(), StorageError> { + use std::{iter::once, os::windows::ffi::OsStrExt, ptr::null_mut}; + use windows_sys::Win32::{ + Foundation::{CloseHandle, INVALID_HANDLE_VALUE}, + Storage::FileSystem::{ + CreateFileW, FILE_FLAG_BACKUP_SEMANTICS, FILE_GENERIC_READ, FILE_GENERIC_WRITE, + FILE_SHARE_DELETE, FILE_SHARE_READ, FILE_SHARE_WRITE, FlushFileBuffers, OPEN_EXISTING, + }, + }; + + let wide_path = path + .as_os_str() + .encode_wide() + .chain(once(0)) + .collect::>(); + // SAFETY: `wide_path` is NUL-terminated and remains valid for this call. + let handle = unsafe { + CreateFileW( + wide_path.as_ptr(), + FILE_GENERIC_READ | FILE_GENERIC_WRITE, + FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, + null_mut(), + OPEN_EXISTING, + FILE_FLAG_BACKUP_SEMANTICS, + null_mut(), + ) + }; + if handle == INVALID_HANDLE_VALUE { + return Err(StorageError::stdio( + IoError::last_os_error().kind(), + "open dir for fsync", + path.display(), + )); + } + + // SAFETY: `handle` is valid until CloseHandle below. + let result = unsafe { FlushFileBuffers(handle) }; + let flush_error = (result == 0).then(IoError::last_os_error); + // SAFETY: CreateFileW returned this owned handle, which is closed exactly once. + let close_result = unsafe { CloseHandle(handle) }; + if let Some(error) = flush_error { + return Err(StorageError::stdio(error.kind(), "fsync dir", path.display())); + } + if close_result == 0 { + return Err(StorageError::stdio( + IoError::last_os_error().kind(), + "close dir after fsync", + path.display(), + )); + } + Ok(()) +} + pub(super) struct PosixReader { path: StoragePath, + #[cfg(unix)] file: Arc, + #[cfg(windows)] + read_file: Arc, file_id: FileId, drop: DeleteOnDrop, @@ -67,6 +170,7 @@ impl Debug for PosixReader { } impl PosixReader { + #[cfg(unix)] fn new( path: StoragePath, file: Arc, @@ -84,6 +188,26 @@ impl PosixReader { ioop_delay, } } + + #[cfg(windows)] + fn new( + path: StoragePath, + file: Arc, + file_id: FileId, + drop: DeleteOnDrop, + async_threads: bool, + ioop_delay: Duration, + ) -> Result { + let read_file = reader_file(&file)?; + Ok(Self { + path, + read_file, + file_id, + drop, + async_threads, + ioop_delay, + }) + } fn open( path: StoragePath, file_name: PathBuf, @@ -100,8 +224,10 @@ impl PosixReader { let size = file .metadata() .map_err(|e| StorageError::stdio(e.kind(), "fstat", file_name.display()))? - .size(); + .len(); + #[cfg(unix)] + { Ok(Arc::new(Self::new( path, Arc::new(file), @@ -110,6 +236,25 @@ impl PosixReader { async_threads, ioop_delay, ))) + } + + #[cfg(windows)] + { + let reader_path = file_name.clone(); + Ok(Arc::new( + Self::new( + path, + Arc::new(file), + FileId::new(), + DeleteOnDrop::new(file_name, true, size, usage), + async_threads, + ioop_delay, + ) + .map_err(|e| { + StorageError::stdio(e.kind(), "open overlapped reader", reader_path.display()) + })?, + )) + } } } @@ -124,6 +269,15 @@ impl FileRw for PosixReader { impl FileCommitter for PosixReader { fn commit(&self) -> Result<(), StorageError> { + #[cfg(windows)] + { + // PosixWriter::complete flushes the writable handle before renaming + // the file into its immutable path. This reader owns a read-only + // overlapped handle, which FlushFileBuffers cannot flush. + return Ok(()); + } + + #[cfg(unix)] self.file .sync_all() .map_err(|e| StorageError::stdio(e.kind(), "fsync", self.drop.path.display())) @@ -141,7 +295,16 @@ impl FileReader for PosixReader { sleep(self.ioop_delay); let mut buffer = FBuf::with_capacity(location.size); - match buffer.read_exact_at(&self.file, location.offset, location.size) { + #[cfg(unix)] + let result = buffer.read_exact_at(&self.file, location.offset, location.size); + #[cfg(windows)] + let result = read_exact_at_overlapped( + &mut buffer, + &self.read_file, + location.offset, + location.size, + ); + match result { Ok(()) => Ok(Arc::new(buffer)), Err(e) => Err(StorageError::stdio( e.kind(), @@ -158,7 +321,10 @@ impl FileReader for PosixReader { callback: Box, StorageError>>) + Send>, ) { if self.async_threads { + #[cfg(unix)] let file = self.file.clone(); + #[cfg(windows)] + let file = self.read_file.clone(); let ioop_delay = self.ioop_delay; let start = Instant::now(); TOKIO.spawn_blocking(move || { @@ -169,7 +335,16 @@ impl FileReader for PosixReader { .map(|location| { READ_BLOCKS_BYTES.record(location.size); let mut buffer = FBuf::with_capacity(location.size); - match buffer.read_exact_at(&file, location.offset, location.size) { + #[cfg(unix)] + let result = buffer.read_exact_at(&file, location.offset, location.size); + #[cfg(windows)] + let result = read_exact_at_overlapped( + &mut buffer, + &file, + location.offset, + location.size, + ); + match result { Ok(()) => Ok(Arc::new(buffer)), Err(e) => Err(StorageError::StdIo { kind: e.kind(), @@ -284,12 +459,14 @@ impl FileWriter for PosixWriter { self.drop.usage.fetch_sub( finalized_path .metadata() - .map_or(0, |metadata| metadata.size() as i64), + .map_or(0, |metadata| metadata.len() as i64), Ordering::Relaxed, ); fs::rename(&self.drop.path, &finalized_path) .map_err(|e| StorageError::stdio(e.kind(), "rename", self.drop.path.display()))?; + #[cfg(unix)] + { Ok(Arc::new(PosixReader::new( self.name, Arc::new(self.file), @@ -298,6 +475,29 @@ impl FileWriter for PosixWriter { self.async_threads, self.ioop_delay, )) as Arc) + } + + #[cfg(windows)] + { + let reader_path = finalized_path.clone(); + Ok(Arc::new( + PosixReader::new( + self.name, + Arc::new(self.file), + self.file_id, + self.drop.with_path(finalized_path), + self.async_threads, + self.ioop_delay, + ) + .map_err(|e| { + StorageError::stdio( + e.kind(), + "open overlapped reader", + reader_path.display(), + ) + })?, + ) as Arc) + } }) } } @@ -441,7 +641,7 @@ impl PosixBackend { if file_type.is_dir() { self.remove_dir_all_recursive(&path) } else if file_type.is_file() { - let size = child.metadata().map_or(0, |metadata| metadata.size()); + let size = child.metadata().map_or(0, |metadata| metadata.len()); fs::remove_file(&path).inspect(|_| { self.usage.fetch_sub(size as i64, Ordering::Relaxed); }) @@ -518,7 +718,7 @@ impl StorageBackend for PosixBackend { .map_err(|e| { StorageError::stdio(e.kind(), "readdir fstat", entry.path().display()) })? - .size(), + .len(), } } else if file_type.is_dir() { StorageFileType::Directory @@ -582,7 +782,7 @@ impl StorageBackend for PosixBackend { .map_err(|e| StorageError::stdio(e.kind(), "unlink", path.display()))?; if metadata.file_type().is_file() { self.usage - .fetch_sub(metadata.size() as i64, Ordering::Relaxed); + .fetch_sub(metadata.len() as i64, Ordering::Relaxed); } Ok(()) } diff --git a/crates/dbsp/src/storage/dirlock.rs b/crates/dbsp/src/storage/dirlock.rs index 70a392359ad..1a39b5e9b99 100644 --- a/crates/dbsp/src/storage/dirlock.rs +++ b/crates/dbsp/src/storage/dirlock.rs @@ -3,11 +3,16 @@ //! Makes sure we don't accidentally run multiple instances of the program //! using the same data directory. +#[cfg(unix)] use libc::{c_int, c_short}; use std::collections::HashSet; -use std::fs::{self, File}; +#[cfg(unix)] +use std::fs; +use std::fs::File; use std::io::{Error as IoError, ErrorKind}; +#[cfg(unix)] use std::os::fd::AsRawFd; +#[cfg(unix)] use std::os::unix::fs::MetadataExt; use std::path::{Path, PathBuf}; use std::process; @@ -31,7 +36,11 @@ mod test; /// Linux has its own "open file description locks", introduced in Linux 3.15, /// that are non-POSIX, which avoid this problem. Maybe we should use those /// instead, if portability and backward compatibility are not paramount. -static LOCKS: LazyLock>> = LazyLock::new(|| Mutex::new(HashSet::new())); +#[cfg(unix)] +type LockId = (u64, u64); +#[cfg(windows)] +type LockId = (u32, u64); +static LOCKS: LazyLock>> = LazyLock::new(|| Mutex::new(HashSet::new())); /// An instance of a PID file. #[derive(Debug)] @@ -48,15 +57,16 @@ pub struct LockedDirectory { /// Device and inode of the file, so that we can remove ourselves from /// [LOCKS] when we're dropped. - dev_ino: (u64, u64), + lock_id: LockId, } impl Drop for LockedDirectory { fn drop(&mut self) { - assert!(LOCKS.lock().unwrap().remove(&self.dev_ino)); + assert!(LOCKS.lock().unwrap().remove(&self.lock_id)); } } +#[cfg(unix)] fn fcntl_lock(file: &File, cmd: c_int) -> Result { let mut flock = libc::flock { l_type: libc::F_WRLCK as c_short, @@ -71,10 +81,12 @@ fn fcntl_lock(file: &File, cmd: c_int) -> Result { } } +#[cfg(unix)] fn write_lock(file: &File) -> Result<(), IoError> { fcntl_lock(file, libc::F_SETLK).map(|_| ()) } +#[cfg(unix)] fn get_lock(file: &File) -> Result, IoError> { // workaround: the size / type of `libc::F_UNLCK` can be different in // different platforms. Resulting in the follow up pattern matching to fail. @@ -91,6 +103,64 @@ fn get_lock(file: &File) -> Result, IoError> { }) } +#[cfg(windows)] +/// Attempts to acquire an exclusive nonblocking lock over the entire file. +fn write_lock(file: &File) -> Result<(), IoError> { + use std::os::windows::io::AsRawHandle; + use windows_sys::Win32::{ + Storage::FileSystem::{LOCKFILE_EXCLUSIVE_LOCK, LOCKFILE_FAIL_IMMEDIATELY, LockFileEx}, + System::IO::OVERLAPPED, + }; + + let mut overlapped = OVERLAPPED::default(); + // SAFETY: `file` owns a valid handle and `overlapped` remains valid for + // this synchronous whole-file lock request. + let result = unsafe { + LockFileEx( + file.as_raw_handle(), + LOCKFILE_EXCLUSIVE_LOCK | LOCKFILE_FAIL_IMMEDIATELY, + 0, + u32::MAX, + u32::MAX, + &mut overlapped, + ) + }; + if result == 0 { + Err(IoError::last_os_error()) + } else { + Ok(()) + } +} + +#[cfg(windows)] +/// Windows does not expose the PID owning a conflicting file lock. +fn get_lock(_file: &File) -> Result, IoError> { + Ok(None) +} + +#[cfg(windows)] +/// Returns a stable identity for `file` from its volume serial number and file index. +fn lock_id(file: &File) -> Result { + use std::{mem::MaybeUninit, os::windows::io::AsRawHandle}; + use windows_sys::Win32::Storage::FileSystem::{ + BY_HANDLE_FILE_INFORMATION, GetFileInformationByHandle, + }; + + let mut info = MaybeUninit::::uninit(); + // SAFETY: `file` owns a valid handle and `info` provides writable storage + // for GetFileInformationByHandle's output. + let result = unsafe { GetFileInformationByHandle(file.as_raw_handle(), info.as_mut_ptr()) }; + if result == 0 { + return Err(IoError::last_os_error()); + } + // SAFETY: GetFileInformationByHandle returned success, initializing `info`. + let info = unsafe { info.assume_init() }; + Ok(( + info.dwVolumeSerialNumber, + (u64::from(info.nFileIndexHigh) << 32) | u64::from(info.nFileIndexLow), + )) +} + impl LockedDirectory { pub const LOCKFILE_NAME: &'static str = "feldera.pidlock"; @@ -116,6 +186,7 @@ impl LockedDirectory { loop { // Did we already lock it? let mut locks = LOCKS.lock().unwrap(); + #[cfg(unix)] match fs::metadata(&pid_file) { Ok(metadata) => { let dev_ino = (metadata.dev(), metadata.ino()); @@ -140,10 +211,20 @@ impl LockedDirectory { .truncate(false) .open(&pid_file) .map_err(|e| StorageError::stdio(e.kind(), "create", pid_file.display()))?; + #[cfg(unix)] let metadata = file .metadata() .map_err(|e| StorageError::stdio(e.kind(), "fstat", pid_file.display()))?; - let dev_ino = (metadata.dev(), metadata.ino()); + #[cfg(unix)] + let lock_id = (metadata.dev(), metadata.ino()); + #[cfg(windows)] + let lock_id = lock_id(&file) + .map_err(|e| StorageError::stdio(e.kind(), "fstat", pid_file.display()))?; + + #[cfg(windows)] + if locks.contains(&lock_id) { + return Err(StorageError::StorageLocked(process::id(), base)); + } match write_lock(&file) { Err(error) @@ -186,11 +267,11 @@ impl LockedDirectory { start.elapsed().as_secs_f64() ); } - locks.insert(dev_ino); + locks.insert(lock_id); return Ok(Self { base, _file: file, - dev_ino, + lock_id, }); } }; diff --git a/crates/dbsp/src/storage/dirlock/test.rs b/crates/dbsp/src/storage/dirlock/test.rs index ae92e73de2d..8fd30d90e68 100644 --- a/crates/dbsp/src/storage/dirlock/test.rs +++ b/crates/dbsp/src/storage/dirlock/test.rs @@ -1,6 +1,10 @@ use super::LockedDirectory; +#[cfg(windows)] +use super::lock_id; use crate::storage::backend::StorageError::{self}; use std::{fs, path::Path, process}; +#[cfg(windows)] +use std::fs::File; fn cant_relock(path: &Path) { let StorageError::StorageLocked(pid, dir) = LockedDirectory::new(path).unwrap_err() else { @@ -63,3 +67,17 @@ fn test_multiple_locks() { drop(_c); drop(_a); } + +#[cfg(windows)] +#[test] +fn lock_id_is_shared_by_hard_links() { + let temp_dir = tempfile::tempdir().unwrap(); + let original_path = temp_dir.path().join("original"); + let hard_link_path = temp_dir.path().join("hard-link"); + fs::write(&original_path, []).unwrap(); + fs::hard_link(&original_path, &hard_link_path).unwrap(); + + let original = File::open(&original_path).unwrap(); + let hard_link = File::open(&hard_link_path).unwrap(); + assert_eq!(lock_id(&original).unwrap(), lock_id(&hard_link).unwrap()); +} diff --git a/crates/samply/Cargo.toml b/crates/samply/Cargo.toml index fa949c5a291..c60bb810198 100644 --- a/crates/samply/Cargo.toml +++ b/crates/samply/Cargo.toml @@ -20,7 +20,6 @@ flate2 = { workspace = true } itertools = { workspace = true } libc = { workspace = true } memory-stats = { workspace = true } -nix = { workspace = true, features = ["process", "time"] } serde = { workspace = true, features = ["derive"] } serde_json = { workspace = true } serde_json_path_to_error = { workspace = true } @@ -30,6 +29,9 @@ thread-id = { workspace = true } tokio = { workspace = true, features = ["sync"] } tracing = { workspace = true } +[target.'cfg(unix)'.dependencies] +nix = { workspace = true, features = ["process", "time"] } + [target.'cfg(target_os = "macos")'.dependencies] mach2 = { workspace = true } diff --git a/crates/samply/src/lib.rs b/crates/samply/src/lib.rs index 7c12ae78cda..32bd0b095b8 100644 --- a/crates/samply/src/lib.rs +++ b/crates/samply/src/lib.rs @@ -52,6 +52,8 @@ use std::{ time::{Duration, Instant}, }; +#[cfg(not(unix))] +use std::sync::OnceLock; #[cfg(target_os = "macos")] use std::time::{SystemTime, UNIX_EPOCH}; @@ -62,7 +64,7 @@ use flate2::{ }; use itertools::Itertools; use memory_stats::memory_stats; -#[cfg(not(target_os = "macos"))] +#[cfg(all(unix, not(target_os = "macos")))] use nix::time::{ClockId, clock_gettime}; use serde::{Deserialize, Serialize}; use serde_json::{Value, json}; @@ -108,14 +110,18 @@ struct Timestamp( /// Monotonic time in nanoseconds. /// /// On macOS this is [`mach_absolute_time`] converted to nanoseconds via - /// [`mach_timebase_info`]. On other Unix platforms this is - /// `CLOCK_MONOTONIC`. + /// [`mach_timebase_info`]. On other Unix platforms this is + /// `CLOCK_MONOTONIC`. On Windows and other non-Unix platforms this is time + /// elapsed since the first timestamp requested by this process. /// /// [`mach_absolute_time`]: https://developer.apple.com/documentation/kernel/1462446-mach_absolute_time /// [`mach_timebase_info`]: https://developer.apple.com/documentation/kernel/1462447-mach_timebase_info i64, ); +#[cfg(not(unix))] +static TIMESTAMP_ORIGIN: OnceLock = OnceLock::new(); + #[cfg(target_os = "macos")] fn mach_absolute_time_nanos() -> i64 { use mach2::mach_time::{mach_absolute_time, mach_timebase_info, mach_timebase_info_data_t}; @@ -144,11 +150,21 @@ impl Timestamp { Self(mach_absolute_time_nanos()) } - #[cfg(not(target_os = "macos"))] + #[cfg(all(unix, not(target_os = "macos")))] { let now = clock_gettime(ClockId::CLOCK_MONOTONIC).unwrap(); Self(now.tv_sec() as i64 * 1_000_000_000 + now.tv_nsec() as i64) } + + #[cfg(not(unix))] + { + Self( + TIMESTAMP_ORIGIN + .get_or_init(Instant::now) + .elapsed() + .as_nanos() as i64, + ) + } } /// Computes `self - other`, returning zero if `self < other`. @@ -172,20 +188,51 @@ fn unix_epoch_nanos() -> i64 { impl From for Timestamp { fn from(value: Instant) -> Self { - // SAFETY: On Unix, `Instant` is implemented using CLOCK_MONOTONIC, - // which is the clock that we need to use for the profiler, but the Rust - // standard library provides no way to get the value out. We don't want - // to make assumptions about the layout of [Instant], and in fact it is - // not defined as libc's struct timespec but different and - // Rust-specific. If we just transmute then we get the wrong value. It - // seems rather safer to assume that the all-bytes-zeros `Instant` is - // the origin, and it works OK for now at least. - // - // The completely safe alternative would be to make Timestamp public and - // force clients to always get both a Timestamp and an Instant if they - // need both, which is wasteful. - let zero = unsafe { std::mem::zeroed::() }; - Self((value - zero).as_nanos() as i64) + #[cfg(not(unix))] + { + let origin = *TIMESTAMP_ORIGIN.get_or_init(Instant::now); + let nanos = match value.checked_duration_since(origin) { + Some(duration) => duration.as_nanos() as i64, + None => -(origin.duration_since(value).as_nanos() as i64), + }; + return Self(nanos); + } + + #[cfg(unix)] + { + // SAFETY: On Unix, `Instant` is implemented using CLOCK_MONOTONIC, + // which is the clock that we need to use for the profiler, but the Rust + // standard library provides no way to get the value out. We don't want + // to make assumptions about the layout of [Instant], and in fact it is + // not defined as libc's struct timespec but different and + // Rust-specific. If we just transmute then we get the wrong value. It + // seems rather safer to assume that the all-bytes-zeros `Instant` is + // the origin, and it works OK for now at least. + // + // The completely safe alternative would be to make Timestamp public and + // force clients to always get both a Timestamp and an Instant if they + // need both, which is wasteful. + let zero = unsafe { std::mem::zeroed::() }; + Self((value - zero).as_nanos() as i64) + } + } +} + +#[cfg(test)] +mod timestamp_tests { + use super::Timestamp; + use std::time::Instant; + + #[cfg(not(unix))] + #[test] + fn instant_conversion_uses_the_timestamp_clock() { + let before = Timestamp::now(); + let instant = Instant::now(); + let converted = Timestamp::from(instant); + let after = Timestamp::now(); + + assert!(before <= converted); + assert!(converted <= after); } } @@ -713,7 +760,7 @@ impl Capture { tooltip, }; markers - .entry(nix::unistd::getpid().as_raw() as usize) + .entry(std::process::id() as usize) .or_default() .1 .0 diff --git a/crates/storage/Cargo.toml b/crates/storage/Cargo.toml index 1047d2c7ea7..0f12ced5ab7 100644 --- a/crates/storage/Cargo.toml +++ b/crates/storage/Cargo.toml @@ -31,3 +31,6 @@ size-of = { workspace = true } [target.'cfg(target_family = "unix")'.dependencies] nix = { version = "0.27.1", features = ["uio", "feature", "fs"] } + +[target.'cfg(windows)'.dependencies] +windows-sys = { version = "0.61", features = ["Win32_Foundation", "Win32_Storage_FileSystem", "Win32_System_IO"] } diff --git a/crates/storage/src/fbuf.rs b/crates/storage/src/fbuf.rs index 6b9fe5037b6..e0b5c8cca01 100644 --- a/crates/storage/src/fbuf.rs +++ b/crates/storage/src/fbuf.rs @@ -13,18 +13,20 @@ use std::{ alloc, borrow::{Borrow, BorrowMut}, - cmp::Ordering, fmt, fs::File, io::{self, Error as IoError, ErrorKind, Read}, mem, ops::{Deref, DerefMut, Index, IndexMut}, - os::fd::AsRawFd, ptr::NonNull, slice, }; -use libc::c_void; +#[cfg(unix)] +use std::os::unix::fs::FileExt; +#[cfg(windows)] +use std::os::windows::io::{AsRawHandle, FromRawHandle}; + use rkyv::{ Archive, Archived, Serialize, ser::{ScratchSpace, Serializer}, @@ -551,42 +553,177 @@ impl FBuf { /// Reads `len` bytes from `file` at the given `offset` and appends them to /// this `FBuf`. /// - /// This avoids zero-initializing the buffer before reading into it. + /// This does not change the file's current position and avoids + /// zero-initializing the buffer before reading into it. + /// + /// Returns an error if fewer than `len` bytes are available or an I/O error + /// occurs. pub fn read_exact_at( &mut self, file: &File, + offset: u64, + len: usize, + ) -> Result<(), IoError> { + #[cfg(unix)] + { + self.read_exact_at_with(offset, len, |buffer, offset| { + read_at(file, buffer, offset) + }) + } + + #[cfg(windows)] + { + let read_file = reopen_for_overlapped_read(file)?; + self.read_exact_at_overlapped(&read_file, offset, len) + } + } + + /// Reads `len` bytes from an overlapped Windows file handle at the given + /// `offset` and appends them to this `FBuf`. + /// + /// The file must have been opened with `FILE_FLAG_OVERLAPPED`. + /// + /// Returns an error if fewer than `len` bytes are available or an I/O error + /// occurs. + #[cfg(windows)] + pub fn read_exact_at_overlapped( + &mut self, + file: &File, + offset: u64, + len: usize, + ) -> Result<(), IoError> { + self.read_exact_at_with(offset, len, |buffer, offset| { + read_at_overlapped(file, buffer, offset) + }) + } + + /// Reads and appends exactly `len` bytes by repeatedly invoking `read_at`. + /// + /// `read_at` must read from the supplied offset without changing the file's + /// current position. Returns an error if it returns zero bytes before the + /// requested length is read or reports an I/O error. + fn read_exact_at_with( + &mut self, mut offset: u64, mut len: usize, - ) -> Result<(), IoError> { + mut read_at: F, + ) -> Result<(), IoError> + where + F: FnMut(&mut [u8], u64) -> Result, + { self.reserve(len); while len > 0 { - let retval = unsafe { - libc::pread( - file.as_raw_fd(), - self.as_mut_ptr().add(self.len) as *mut c_void, - len, - offset as i64, - ) + // SAFETY: `reserve` ensures that the allocation has room for `len` + // bytes after `self.len`. The exclusive `&mut self` borrow prevents + // reallocation while `read_at` uses the returned slice. + let read_buf = + unsafe { slice::from_raw_parts_mut(self.as_mut_ptr().add(self.len), len) }; + let read = match read_at(read_buf, offset) { + Err(error) if error.kind() == ErrorKind::Interrupted => continue, + result => result?, }; - match retval.cmp(&0) { - Ordering::Equal => return Err(ErrorKind::UnexpectedEof.into()), - Ordering::Less => { - let error = IoError::last_os_error(); - if error.kind() != ErrorKind::Interrupted { - return Err(error); - } - } - Ordering::Greater => { - self.len += retval as usize; - len -= retval as usize; - offset += retval as u64; - } + if read == 0 { + return Err(ErrorKind::UnexpectedEof.into()); } + self.len += read; + len -= read; + offset += read as u64; } Ok(()) } } +#[cfg(unix)] +/// Reads from `file` at `offset` without changing the file's current position. +fn read_at(file: &File, buffer: &mut [u8], offset: u64) -> Result { + file.read_at(buffer, offset) +} + +#[cfg(windows)] +/// Reopens `file` with overlapped I/O enabled for positioned reads. +fn reopen_for_overlapped_read(file: &File) -> Result { + // The caller's file may not have FILE_FLAG_OVERLAPPED. The returned File + // owns the duplicated handle used for positioned reads. + use windows_sys::Win32::{ + Foundation::INVALID_HANDLE_VALUE, + Storage::FileSystem::{ + FILE_FLAG_OVERLAPPED, FILE_GENERIC_READ, FILE_SHARE_DELETE, FILE_SHARE_READ, + FILE_SHARE_WRITE, ReOpenFile, + }, + }; + + // SAFETY: `file` owns a valid handle, which remains borrowed for the + // duration of this call. + let handle = unsafe { + ReOpenFile( + file.as_raw_handle(), + FILE_GENERIC_READ, + FILE_SHARE_READ | FILE_SHARE_WRITE | FILE_SHARE_DELETE, + FILE_FLAG_OVERLAPPED, + ) + }; + if handle == INVALID_HANDLE_VALUE { + return Err(IoError::last_os_error()); + } + + // SAFETY: `ReOpenFile` returned a distinct owned handle after the invalid + // handle case was excluded. The returned `File` takes ownership of it. + Ok(unsafe { File::from_raw_handle(handle) }) +} + +#[cfg(windows)] +/// Reads from an overlapped `file` at `offset` and waits for the operation to complete. +fn read_at_overlapped(file: &File, buffer: &mut [u8], offset: u64) -> Result { + use windows_sys::Win32::{ + Foundation::ERROR_IO_PENDING, + Storage::FileSystem::ReadFile, + System::IO::{GetOverlappedResult, OVERLAPPED, OVERLAPPED_0_0}, + }; + + let byte_count = u32::try_from(buffer.len()).map_err(|_| { + IoError::new( + ErrorKind::InvalidInput, + "read buffer length exceeds the Windows ReadFile limit", + ) + })?; + let mut overlapped = OVERLAPPED::default(); + overlapped.Anonymous.Anonymous = OVERLAPPED_0_0 { + Offset: offset as u32, + OffsetHigh: (offset >> 32) as u32, + }; + let mut bytes_read = 0; + + // SAFETY: `buffer` is writable for `byte_count` bytes, `file` owns a valid + // handle, and `overlapped` is initialized for this operation. + let result = unsafe { + ReadFile( + file.as_raw_handle(), + buffer.as_mut_ptr(), + byte_count, + &mut bytes_read, + &mut overlapped, + ) + }; + if result == 0 { + let error = IoError::last_os_error(); + if error.raw_os_error() != Some(ERROR_IO_PENDING as i32) { + return Err(error); + } + + // Wait for pending I/O before returning so `overlapped` and `buffer` remain valid. + // SAFETY: This uses the same handle and `OVERLAPPED` as the pending + // `ReadFile` operation and waits for it to complete before either is dropped. + if unsafe { + GetOverlappedResult(file.as_raw_handle(), &mut overlapped, &mut bytes_read, 1) + } == 0 + { + return Err(IoError::last_os_error()); + } + } + + Ok(bytes_read as usize) +} + impl From for Vec { #[inline] fn from(aligned: FBuf) -> Self { @@ -870,3 +1007,72 @@ impl + BorrowMut> Serializer for FBufSerializer { } } } + +#[cfg(test)] +mod tests { + use super::FBuf; + use std::{ + fs::{File, OpenOptions}, + io::{Read, Seek, SeekFrom, Write}, + path::PathBuf, + time::{SystemTime, UNIX_EPOCH}, + }; + #[cfg(windows)] + use std::os::windows::fs::OpenOptionsExt; + + fn temporary_file_path() -> PathBuf { + std::env::temp_dir().join(format!( + "feldera-storage-fbuf-read-at-{}", + SystemTime::now() + .duration_since(UNIX_EPOCH) + .unwrap() + .as_nanos() + )) + } + + #[test] + fn read_exact_at_reads_from_offset_without_seeking_file() { + let path = temporary_file_path(); + let mut writer = File::create(&path).unwrap(); + writer.write_all(b"0123456789abcdef").unwrap(); + drop(writer); + + let mut file = OpenOptions::new().read(true).open(&path).unwrap(); + file.seek(SeekFrom::Start(2)).unwrap(); + + let mut buffer = FBuf::new(); + buffer.read_exact_at(&file, 8, 4).unwrap(); + assert_eq!(buffer.as_slice(), b"89ab"); + + let mut next_byte = [0_u8; 1]; + file.read_exact(&mut next_byte).unwrap(); + assert_eq!(next_byte, *b"2"); + + drop(file); + std::fs::remove_file(path).unwrap(); + } + + #[cfg(windows)] + #[test] + fn read_exact_at_overlapped_reads_from_offset() { + use windows_sys::Win32::Storage::FileSystem::FILE_FLAG_OVERLAPPED; + + let path = temporary_file_path(); + let mut writer = File::create(&path).unwrap(); + writer.write_all(b"0123456789abcdef").unwrap(); + drop(writer); + + let file = OpenOptions::new() + .read(true) + .custom_flags(FILE_FLAG_OVERLAPPED) + .open(&path) + .unwrap(); + + let mut buffer = FBuf::new(); + buffer.read_exact_at_overlapped(&file, 8, 4).unwrap(); + assert_eq!(buffer.as_slice(), b"89ab"); + + drop(file); + std::fs::remove_file(path).unwrap(); + } +}