Skip to content

Windows compatibility for dbsp and dependencies #6627

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
vpopescu wants to merge 1 commit into from
Closed

Windows compatibility for dbsp and dependencies #6627

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4,698 changes: 2,365 additions & 2,333 deletions Cargo.lock
View file
Open in desktop

Large diffs are not rendered by default.

11 changes: 9 additions & 2 deletions crates/dbsp/Cargo.toml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 }
Expand All @@ -115,14 +120,16 @@ 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"] }
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
Expand Down
10 changes: 10 additions & 0 deletions crates/dbsp/benches/column_layer.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -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);
69 changes: 61 additions & 8 deletions crates/dbsp/src/circuit/circuit_builder.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -59,6 +59,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},
Expand Down Expand Up @@ -7984,26 +7985,69 @@ impl<T> Timed<T> {
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)]
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)]
fn current_thread_cpu_time() -> Duration {

@mihaibudiu mihaibudiu Jul 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It sure looks hard to read a timer in Windows.

@vpopescu vpopescu Jul 13, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's really just an extra step conversion step.

use std::mem::MaybeUninit;
use windows_sys::Win32::{
Foundation::FILETIME,
System::Threading::{GetCurrentThread, GetThreadTimes},
};

let mut creation = MaybeUninit::<FILETIME>::uninit();
let mut exit = MaybeUninit::<FILETIME>::uninit();
let mut kernel = MaybeUninit::<FILETIME>::uninit();
let mut user = MaybeUninit::<FILETIME>::uninit();
let result = unsafe {
GetThreadTimes(
GetCurrentThread(),
creation.as_mut_ptr(),
exit.as_mut_ptr(),
kernel.as_mut_ptr(),
user.as_mut_ptr(),
)
};
// this is an almost impossible error, but we should panic if it happens
if result == 0 {
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(unsafe { kernel.assume_init() })
.saturating_add(filetime_to_ticks(unsafe { user.assume_init() }));
Duration::new(ticks / 10_000_000, ((ticks % 10_000_000) * 100) as u32)
}

#[cfg(not(any(unix, windows)))]
fn current_thread_cpu_time() -> Duration {

@mihaibudiu mihaibudiu Jul 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

If this is safe, perhaps it's safe to return 0 in windows if you can't read the timers as well
Alternatively, this can be unimplemented!()

@vpopescu vpopescu Jul 13, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Unimplemented!() would just rob us of a feature that's unlikely to fail in normal operation. I can certainly return "0" but I have not tracked up the code to see how that is used.

@vpopescu vpopescu Jul 13, 2026 •
edited
Loading

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Looking further, I can return a 0, but will cause incorrect profiling results, so I'm not sure how you feel about that. I traced it a bit further, and we could do it, but if I missed something and it's used in some kind of math at some point, it may trigger additional issues.

A better fix may be to return an Option<> so we can indicate to the caller that the result is invalid, but this means more changes downstream. Another option would be to consider '0' as invalid, but a newly created thread can legitimately report a time of 0.

I'd leave it as is since it only affects windows, and the likelihood of failure is not high, and if panic!() ever gets triggered, It's really about the same failure chance as the clock_gettime() in clock_gettime(),

@vpopescu vpopescu Jul 13, 2026

Copy link
Copy Markdown
Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Eh, i asked the all known AI:

  1. GetCurrentThread() Cannot Fail
    The comment speculates that GetCurrentThread() might return a "bad handle," but this is impossible by design.

GetCurrentThread() does not open a real handle or allocate OS resources. It simply returns a hardcoded pseudo-handle (which is always the constant value (HANDLE)-2).

Because it is a pseudo-handle, it is guaranteed to be valid for the calling thread and inherently possesses the THREAD_QUERY_INFORMATION access rights required by GetThreadTimes. It cannot be closed by mistake, and it cannot be "bad."

  1. The Pointers are Guaranteed to be Valid
    The API requires valid memory addresses to write the time data into. Assuming creation, exit, kernel, and user are standard Rust local variables (like MaybeUninit), calling as_mut_ptr() on them yields valid, non-null pointers to your thread's stack.

The Windows kernel checks if it can write to these pointers. Since they point to memory safely managed by Rust's compiler, the memory is guaranteed to be allocated and writable.

When Could It Theoretically Fail?
The only way GetThreadTimes returns 0 in this exact snippet is if your program has already suffered catastrophic Undefined Behavior (UB) elsewhere.

@mihaibudiu mihaibudiu Jul 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ok, leave it as it is then. Thank you.

@mihaibudiu mihaibudiu Jul 13, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Then panic is the right choice

Duration::ZERO
}

impl<T> Future for Timed<T>
where
T: Future,
Expand All @@ -8023,6 +8067,7 @@ where

#[cfg(test)]
mod tests {

use crate::{
Circuit, Error as DbspError, RootCircuit,
circuit::schedule::{DynamicScheduler, Scheduler},
Expand All @@ -8032,6 +8077,14 @@ 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::<DynamicScheduler>();
Expand Down
2 changes: 2 additions & 0 deletions crates/dbsp/src/storage/backend.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
}
}
Loading