From d0ca43c1fb5c658fd90b73f4c6eacea3428eaf13 Mon Sep 17 00:00:00 2001 From: Val <7304633+vpopescu@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:07:55 -0500 Subject: [PATCH 1/3] [storage] Updated storage for windows compatibility Replaced the Unix-only pread implementation with platform-specific positioned reads. Unix uses FileExt::read_at, while Windows duplicates an overlapped file handle once per operation and reads with ReadFile/GetOverlappedResult. Unlike previous version of this code (abandoned PR), So it opens one duplicated, overlapped handle for the full read_exact_at request and reuses it through every loop iteration. Added regression tests for preserving the file cursor and for Windows overlapped reads. Added the Windows-only windows-sys dependency. --- Cargo.lock | 1 + crates/storage/Cargo.toml | 3 + crates/storage/src/fbuf.rs | 256 +++++++++++++++++++++++++++++++++---- 3 files changed, 235 insertions(+), 25 deletions(-) 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/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(); + } +} From 6947731d903fbe928db75029f824432af40fcedd Mon Sep 17 00:00:00 2001 From: Val <7304633+vpopescu@users.noreply.github.com> Date: Fri, 17 Jul 2026 13:24:46 -0500 Subject: [PATCH 2/3] [timely] Support profiling annotations on Windows Limit nix to Unix targets and use std::process::id for the profiling marker process ID. Generate non-Unix timestamps from a shared process-relative Instant origin. Convert externally supplied Instant span starts using that same origin, keeping with_start markers aligned with captured timestamps. Add a regression test for the conversion. --- crates/samply/Cargo.toml | 4 +- crates/samply/src/lib.rs | 85 +++++++++++++++++++++++++++++++--------- 2 files changed, 69 insertions(+), 20 deletions(-) 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 From 62c67b9617ded145a266a122e6efc3b72e482bc3 Mon Sep 17 00:00:00 2001 From: Val <7304633+vpopescu@users.noreply.github.com> Date: Fri, 17 Jul 2026 15:23:25 -0500 Subject: [PATCH 3/3] [dbsp] Windows compat circuit and storage modules Use GetThreadTimes to measure current-thread CPU time on Windows and retain the existing CLOCK_THREAD_CPUTIME_ID implementation on Unix. Make DBSP storage work on Windows with a reusable overlapped read handle per reader, portable file-length APIs, flushed directory metadata, and `LockFileEx` directory locks keyed by volume serial number and file index. Keep Unix positioned reads, fcntl locking, resource use, and runtime behavior unchanged. Use more portable functions (e.g. len() instead of size() in a few places) Add Windows-only tests for lock identities across hard links, and run the column-layer benchmark without Unix-only flamegraph profiling. --- crates/dbsp/Cargo.toml | 11 +- crates/dbsp/benches/column_layer.rs | 10 + crates/dbsp/src/circuit/circuit_builder.rs | 76 +++++- crates/dbsp/src/storage/backend.rs | 2 + .../dbsp/src/storage/backend/posixio_impl.rs | 216 +++++++++++++++++- crates/dbsp/src/storage/dirlock.rs | 95 +++++++- crates/dbsp/src/storage/dirlock/test.rs | 18 ++ 7 files changed, 400 insertions(+), 28 deletions(-) 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()); +}