Skip to content

Commit

Permalink
Windows: Implement mutex using futex
Browse files Browse the repository at this point in the history
Well, the Windows equivalent: `WaitOnAddress`, `WakeByAddressSingle` and `WakeByAddressAll`.
  • Loading branch information
ChrisDenton committed Mar 3, 2024
1 parent 9e73597 commit 508a095
Show file tree
Hide file tree
Showing 16 changed files with 171 additions and 104 deletions.
6 changes: 6 additions & 0 deletions library/std/src/sys/locks/condvar/futex.rs
Original file line number Diff line number Diff line change
@@ -1,4 +1,10 @@
use crate::sync::atomic::{AtomicU32, Ordering::Relaxed};
#[cfg(windows)]
use crate::sys::api::{
wait_on_address as futex_wait, wake_by_address_all as futex_wake_all,
wake_by_address_single as futex_wake,
};
#[cfg(not(windows))]
use crate::sys::futex::{futex_wait, futex_wake, futex_wake_all};
use crate::sys::locks::Mutex;
use crate::time::Duration;
Expand Down
7 changes: 4 additions & 3 deletions library/std/src/sys/locks/condvar/mod.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
cfg_if::cfg_if! {
if #[cfg(any(
all(target_os = "windows", not(target_vendor="win7")),
target_os = "linux",
target_os = "android",
target_os = "freebsd",
Expand All @@ -14,9 +15,9 @@ cfg_if::cfg_if! {
} else if #[cfg(target_family = "unix")] {
mod pthread;
pub use pthread::Condvar;
} else if #[cfg(target_os = "windows")] {
mod windows;
pub use windows::Condvar;
} else if #[cfg(all(target_os = "windows", target_vendor = "win7"))] {
mod windows7;
pub use windows7::Condvar;
} else if #[cfg(all(target_vendor = "fortanix", target_env = "sgx"))] {
mod sgx;
pub use sgx::Condvar;
Expand Down
101 changes: 7 additions & 94 deletions library/std/src/sys/locks/mutex/futex.rs
Original file line number Diff line number Diff line change
@@ -1,96 +1,9 @@
use crate::sync::atomic::{
AtomicU32,
Ordering::{Acquire, Relaxed, Release},
};
use crate::sys::futex::{futex_wait, futex_wake};

pub struct Mutex {
/// 0: unlocked
/// 1: locked, no other threads waiting
/// 2: locked, and other threads waiting (contended)
futex: AtomicU32,
}

impl Mutex {
#[inline]
pub const fn new() -> Self {
Self { futex: AtomicU32::new(0) }
}

#[inline]
pub fn try_lock(&self) -> bool {
self.futex.compare_exchange(0, 1, Acquire, Relaxed).is_ok()
}

#[inline]
pub fn lock(&self) {
if self.futex.compare_exchange(0, 1, Acquire, Relaxed).is_err() {
self.lock_contended();
}
}

#[cold]
fn lock_contended(&self) {
// Spin first to speed things up if the lock is released quickly.
let mut state = self.spin();

// If it's unlocked now, attempt to take the lock
// without marking it as contended.
if state == 0 {
match self.futex.compare_exchange(0, 1, Acquire, Relaxed) {
Ok(_) => return, // Locked!
Err(s) => state = s,
}
}

loop {
// Put the lock in contended state.
// We avoid an unnecessary write if it as already set to 2,
// to be friendlier for the caches.
if state != 2 && self.futex.swap(2, Acquire) == 0 {
// We changed it from 0 to 2, so we just successfully locked it.
return;
}

// Wait for the futex to change state, assuming it is still 2.
futex_wait(&self.futex, 2, None);

// Spin again after waking up.
state = self.spin();
}
}

fn spin(&self) -> u32 {
let mut spin = 100;
loop {
// We only use `load` (and not `swap` or `compare_exchange`)
// while spinning, to be easier on the caches.
let state = self.futex.load(Relaxed);

// We stop spinning when the mutex is unlocked (0),
// but also when it's contended (2).
if state != 1 || spin == 0 {
return state;
}

crate::hint::spin_loop();
spin -= 1;
}
}

#[inline]
pub unsafe fn unlock(&self) {
if self.futex.swap(0, Release) == 2 {
// We only wake up one thread. When that thread locks the mutex, it
// will mark the mutex as contended (2) (see lock_contended above),
// which makes sure that any other waiting threads will also be
// woken up eventually.
self.wake();
}
}

#[cold]
fn wake(&self) {
futex_wake(&self.futex);
cfg_if::cfg_if! {
if #[cfg(windows)] {
mod windows;
pub use windows::*;
} else {
mod unix;
pub use unix::*;
}
}
1 change: 1 addition & 0 deletions library/std/src/sys/locks/mutex/futex/unix.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1 @@

65 changes: 65 additions & 0 deletions library/std/src/sys/locks/mutex/futex/windows.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,65 @@
use crate::sync::atomic::{
AtomicI8,
Ordering::{Acquire, Relaxed, Release},
};
use crate::sys::api;

pub struct Mutex {
state: AtomicI8,
}

const UNLOCKED: i8 = 0;
const LOCKED: i8 = 1;
const CONTENDED: i8 = 2;

impl Mutex {
#[inline]
pub const fn new() -> Mutex {
Mutex { state: AtomicI8::new(UNLOCKED) }
}

#[inline]
pub fn lock(&self) {
if let Err(state) = self.state.compare_exchange(UNLOCKED, LOCKED, Acquire, Relaxed) {
self.lock_contended(state)
}
}

#[cold]
fn lock_contended(&self, mut state: i8) {
// Note: WaitOnAddress is already quite spin-happy so we don't do any further spinning on top.
loop {
// Put the lock in contended state.
// We avoid an unnecessary write if it as already set to CONTENDED,
// to be friendlier for the caches.
if state != CONTENDED && self.state.swap(CONTENDED, Acquire) == UNLOCKED {
// We changed it from UNLOCKED to CONTENDED, so we just successfully locked it.
return;
}
// Wait for the futex to change state, assuming it is still CONTENDED.
api::wait_on_address(&self.state, CONTENDED, None);
state = self.state.load(Relaxed);
}
}

#[inline]
pub fn try_lock(&self) -> bool {
self.state.compare_exchange(UNLOCKED, LOCKED, Acquire, Relaxed).is_ok()
}

#[inline]
pub unsafe fn unlock(&self) {
if self.state.swap(UNLOCKED, Release) == CONTENDED {
// We only wake up one thread. When that thread locks the mutex, it
// will mark the mutex as CONTENDED (see lock_contended above),
// which makes sure that any other waiting threads will also be
// woken up eventually.
self.wake();
}
}

#[cold]
fn wake(&self) {
api::wake_by_address_single(&self.state);
}
}
7 changes: 4 additions & 3 deletions library/std/src/sys/locks/mutex/mod.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
cfg_if::cfg_if! {
if #[cfg(any(
all(target_os = "windows", not(target_vendor = "win7")),
target_os = "linux",
target_os = "android",
target_os = "freebsd",
Expand All @@ -19,9 +20,9 @@ cfg_if::cfg_if! {
))] {
mod pthread;
pub use pthread::{Mutex, raw};
} else if #[cfg(target_os = "windows")] {
mod windows;
pub use windows::{Mutex, raw};
} else if #[cfg(all(target_os = "windows", target_vendor = "win7"))] {
mod windows7;
pub use windows7::{Mutex, raw};
} else if #[cfg(all(target_vendor = "fortanix", target_env = "sgx"))] {
mod sgx;
pub use sgx::Mutex;
Expand Down
File renamed without changes.
15 changes: 15 additions & 0 deletions library/std/src/sys/locks/rwlock/futex.rs
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,12 @@ use crate::sync::atomic::{
AtomicU32,
Ordering::{Acquire, Relaxed, Release},
};
#[cfg(windows)]
use crate::sys::api::{
wait_on_address as futex_wait, wake_by_address_all as futex_wake_all,
wake_by_address_single as futex_wake,
};
#[cfg(not(windows))]
use crate::sys::futex::{futex_wait, futex_wake, futex_wake_all};

pub struct RwLock {
Expand Down Expand Up @@ -291,7 +297,10 @@ impl RwLock {
}

/// Spin for a while, but stop directly at the given condition.
///
/// We avoid spinning on Windows because the futex implementation spins enough.
#[inline]
#[cfg(not(windows))]
fn spin_until(&self, f: impl Fn(u32) -> bool) -> u32 {
let mut spin = 100; // Chosen by fair dice roll.
loop {
Expand All @@ -304,6 +313,12 @@ impl RwLock {
}
}

#[inline]
#[cfg(windows)]
fn spin_until(&self, _f: impl Fn(u32) -> bool) -> u32 {
self.state.load(Relaxed)
}

#[inline]
fn spin_write(&self) -> u32 {
// Stop spinning when it's unlocked or when there's waiting writers, to keep things somewhat fair.
Expand Down
7 changes: 4 additions & 3 deletions library/std/src/sys/locks/rwlock/mod.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
cfg_if::cfg_if! {
if #[cfg(any(
all(target_os = "windows", not(target_vendor = "win7")),
target_os = "linux",
target_os = "android",
target_os = "freebsd",
Expand All @@ -14,9 +15,9 @@ cfg_if::cfg_if! {
} else if #[cfg(target_family = "unix")] {
mod queue;
pub use queue::RwLock;
} else if #[cfg(target_os = "windows")] {
mod windows;
pub use windows::RwLock;
} else if #[cfg(all(target_os = "windows", target_vendor = "win7"))] {
mod windows7;
pub use windows7::RwLock;
} else if #[cfg(all(target_vendor = "fortanix", target_env = "sgx"))] {
mod sgx;
pub use sgx::RwLock;
Expand Down
File renamed without changes.
39 changes: 39 additions & 0 deletions library/std/src/sys/pal/windows/api.rs
Original file line number Diff line number Diff line change
Expand Up @@ -155,3 +155,42 @@ pub fn get_last_error() -> WinError {
pub struct WinError {
pub code: u32,
}

#[cfg(not(target_vendor = "win7"))]
mod futex {
use super::*;
use crate::sys::dur2timeout;
use core::mem;
use core::time::Duration;

#[inline(always)]
pub fn wait_on_address<T, U>(address: &T, compare: U, timeout: Option<Duration>) -> bool {
assert_eq!(mem::size_of::<T>(), mem::size_of::<U>());
unsafe {
let addr = addr_of!(*address).cast::<c_void>();
let size = mem::size_of::<T>();
let compare_addr = addr_of!(compare).cast::<c_void>();
let timeout = timeout.map(dur2timeout).unwrap_or(c::INFINITE);
c::WaitOnAddress(addr, compare_addr, size, timeout) == c::TRUE
}
}

#[inline(always)]
pub fn wake_by_address_single<T>(address: &T) -> bool {
unsafe {
let addr = addr_of!(*address).cast::<c_void>();
c::WakeByAddressSingle(addr);
false
}
}

#[inline(always)]
pub fn wake_by_address_all<T>(address: &T) {
unsafe {
let addr = addr_of!(*address).cast::<c_void>();
c::WakeByAddressAll(addr);
}
}
}
#[cfg(not(target_vendor = "win7"))]
pub use futex::*;
4 changes: 4 additions & 0 deletions library/std/src/sys/pal/windows/c.rs
Original file line number Diff line number Diff line change
Expand Up @@ -36,6 +36,7 @@ pub type LPVOID = *mut c_void;
pub type LPWCH = *mut WCHAR;
pub type LPWSTR = *mut WCHAR;

#[cfg(target_vendor = "win7")]
pub type PSRWLOCK = *mut SRWLOCK;

pub type socklen_t = c_int;
Expand All @@ -50,7 +51,9 @@ pub const INVALID_HANDLE_VALUE: HANDLE = ::core::ptr::without_provenance_mut(-1i
pub const EXIT_SUCCESS: u32 = 0;
pub const EXIT_FAILURE: u32 = 1;

#[cfg(target_vendor = "win7")]
pub const CONDITION_VARIABLE_INIT: CONDITION_VARIABLE = CONDITION_VARIABLE { Ptr: ptr::null_mut() };
#[cfg(target_vendor = "win7")]
pub const SRWLOCK_INIT: SRWLOCK = SRWLOCK { Ptr: ptr::null_mut() };
pub const INIT_ONCE_STATIC_INIT: INIT_ONCE = INIT_ONCE { Ptr: ptr::null_mut() };

Expand Down Expand Up @@ -373,6 +376,7 @@ extern "system" {
dwmilliseconds: u32,
) -> BOOL;
pub fn WakeByAddressSingle(address: *const c_void);
pub fn WakeByAddressAll(address: *const c_void);
}

#[cfg(target_vendor = "win7")]
Expand Down
2 changes: 1 addition & 1 deletion library/std/src/sys/pal/windows/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -39,7 +39,7 @@ cfg_if::cfg_if! {
}
}

mod api;
pub(in crate::sys) mod api;

/// Map a Result<T, WinError> to io::Result<T>.
trait IoResult<T> {
Expand Down
6 changes: 6 additions & 0 deletions src/tools/miri/src/shims/windows/foreign_items.rs
Original file line number Diff line number Diff line change
Expand Up @@ -355,6 +355,12 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {

this.WakeByAddressSingle(ptr_op)?;
}
"WakeByAddressAll" => {
let [ptr_op] =
this.check_shim(abi, Abi::System { unwind: false }, link_name, args)?;

this.WakeByAddressAll(ptr_op)?;
}

// Dynamic symbol loading
"GetProcAddress" => {
Expand Down
15 changes: 15 additions & 0 deletions src/tools/miri/src/shims/windows/sync.rs
Original file line number Diff line number Diff line change
Expand Up @@ -384,6 +384,21 @@ pub trait EvalContextExt<'mir, 'tcx: 'mir>: crate::MiriInterpCxExt<'mir, 'tcx> {

Ok(())
}
fn WakeByAddressAll(&mut self, ptr_op: &OpTy<'tcx, Provenance>) -> InterpResult<'tcx> {
let this = self.eval_context_mut();

let ptr = this.read_pointer(ptr_op)?;

// See the Linux futex implementation for why this fence exists.
this.atomic_fence(AtomicFenceOrd::SeqCst)?;

while let Some(thread) = this.futex_wake(ptr.addr().bytes(), u32::MAX) {
this.unblock_thread(thread);
this.unregister_timeout_callback_if_exists(thread);
}

Ok(())
}

fn SleepConditionVariableSRW(
&mut self,
Expand Down

0 comments on commit 508a095

Please sign in to comment.