OpenE2K
/
rust
2
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Auto merge of #76919 - fusion-engineering-forks:thread-parker, r=dtolnay 

Use futex-based thread::park/unpark on Linux.

This moves the parking/unparking logic out of `thread/mod.rs` into a module named `thread_parker` in `sys_common`. The current implementation is moved to `sys_common/thread_parker/generic.rs` and the new implementation using futexes is added in `sys_common/thread_parker/futex.rs`.
This commit is contained in:
bors 2020-10-01 13:21:34 +00:00
parent 9cba260df0 0b73fd7105
commit 782013564e
7 changed files with 276 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
library/std/src/sys/unix/futex.rs Normal file
View File

@ -0,0 +1,37 @@
#![cfg(any(target_os = "linux", target_os = "android"))]
use crate::convert::TryInto;
use crate::ptr::null;
use crate::sync::atomic::AtomicI32;
use crate::time::Duration;
pub fn futex_wait(futex: &AtomicI32, expected: i32, timeout: Option<Duration>) {
let timespec = timeout.and_then(|d| {
Some(libc::timespec {
// Sleep forever if the timeout is longer than fits in a timespec.
tv_sec: d.as_secs().try_into().ok()?,
// This conversion never truncates, as subsec_nanos is always <1e9.
tv_nsec: d.subsec_nanos() as _,
})
});
unsafe {
libc::syscall(
libc::SYS_futex,
futex as *const AtomicI32,
libc::FUTEX_WAIT | libc::FUTEX_PRIVATE_FLAG,
expected,
timespec.as_ref().map_or(null(), |d| d as *const libc::timespec),
);
}
}
pub fn futex_wake(futex: &AtomicI32) {
unsafe {
libc::syscall(
libc::SYS_futex,
futex as *const AtomicI32,
libc::FUTEX_WAKE | libc::FUTEX_PRIVATE_FLAG,
1,
);
}
}

1
library/std/src/sys/unix/mod.rs
View File

@ -49,6 +49,7 @@ pub mod env;
pub mod ext;
pub mod fd;
pub mod fs;
pub mod futex;
pub mod io;
#[cfg(target_os = "l4re")]
mod l4re;

1
library/std/src/sys_common/mod.rs
View File

@ -66,6 +66,7 @@ pub mod thread;
pub mod thread_info;
pub mod thread_local_dtor;
pub mod thread_local_key;
pub mod thread_parker;
pub mod util;
pub mod wtf8;

93
library/std/src/sys_common/thread_parker/futex.rs Normal file
View File

@ -0,0 +1,93 @@
use crate::sync::atomic::AtomicI32;
use crate::sync::atomic::Ordering::{Acquire, Release};
use crate::sys::futex::{futex_wait, futex_wake};
use crate::time::Duration;
const PARKED: i32 = -1;
const EMPTY: i32 = 0;
const NOTIFIED: i32 = 1;
pub struct Parker {
state: AtomicI32,
}
// Notes about memory ordering:
//
// Memory ordering is only relevant for the relative ordering of operations
// between different variables. Even Ordering::Relaxed guarantees a
// monotonic/consistent order when looking at just a single atomic variable.
//
// So, since this parker is just a single atomic variable, we only need to look
// at the ordering guarantees we need to provide to the 'outside world'.
//
// The only memory ordering guarantee that parking and unparking provide, is
// that things which happened before unpark() are visible on the thread
// returning from park() afterwards. Otherwise, it was effectively unparked
// before unpark() was called while still consuming the 'token'.
//
// In other words, unpark() needs to synchronize with the part of park() that
// consumes the token and returns.
//
// This is done with a release-acquire synchronization, by using
// Ordering::Release when writing NOTIFIED (the 'token') in unpark(), and using
// Ordering::Acquire when checking for this state in park().
impl Parker {
#[inline]
pub const fn new() -> Self {
Parker { state: AtomicI32::new(EMPTY) }
}
// Assumes this is only called by the thread that owns the Parker,
// which means that `self.state != PARKED`.
pub unsafe fn park(&self) {
// Change NOTIFIED=>EMPTY or EMPTY=>PARKED, and directly return in the
// first case.
if self.state.fetch_sub(1, Acquire) == NOTIFIED {
return;
}
loop {
// Wait for something to happen, assuming it's still set to PARKED.
futex_wait(&self.state, PARKED, None);
// Change NOTIFIED=>EMPTY and return in that case.
if self.state.compare_and_swap(NOTIFIED, EMPTY, Acquire) == NOTIFIED {
return;
} else {
// Spurious wake up. We loop to try again.
}
}
}
// Assumes this is only called by the thread that owns the Parker,
// which means that `self.state != PARKED`.
pub unsafe fn park_timeout(&self, timeout: Duration) {
// Change NOTIFIED=>EMPTY or EMPTY=>PARKED, and directly return in the
// first case.
if self.state.fetch_sub(1, Acquire) == NOTIFIED {
return;
}
// Wait for something to happen, assuming it's still set to PARKED.
futex_wait(&self.state, PARKED, Some(timeout));
// This is not just a store, because we need to establish a
// release-acquire ordering with unpark().
if self.state.swap(EMPTY, Acquire) == NOTIFIED {
// Woke up because of unpark().
} else {
// Timeout or spurious wake up.
// We return either way, because we can't easily tell if it was the
// timeout or not.
}
}
#[inline]
pub fn unpark(&self) {
// Change PARKED=>NOTIFIED, EMPTY=>NOTIFIED, or NOTIFIED=>NOTIFIED, and
// wake the thread in the first case.
//
// Note that even NOTIFIED=>NOTIFIED results in a write. This is on
// purpose, to make sure every unpark() has a release-acquire ordering
// with park().
if self.state.swap(NOTIFIED, Release) == PARKED {
futex_wake(&self.state);
}
}
}

119
library/std/src/sys_common/thread_parker/generic.rs Normal file
View File

@ -0,0 +1,119 @@
//! Parker implementaiton based on a Mutex and Condvar.
use crate::sync::atomic::AtomicUsize;
use crate::sync::atomic::Ordering::SeqCst;
use crate::sync::{Condvar, Mutex};
use crate::time::Duration;
const EMPTY: usize = 0;
const PARKED: usize = 1;
const NOTIFIED: usize = 2;
pub struct Parker {
state: AtomicUsize,
lock: Mutex<()>,
cvar: Condvar,
}
impl Parker {
pub fn new() -> Self {
Parker { state: AtomicUsize::new(EMPTY), lock: Mutex::new(()), cvar: Condvar::new() }
}
// This implementaiton doesn't require `unsafe`, but other implementations
// may assume this is only called by the thread that owns the Parker.
pub unsafe fn park(&self) {
// If we were previously notified then we consume this notification and
// return quickly.
if self.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst).is_ok() {
return;
}
// Otherwise we need to coordinate going to sleep
let mut m = self.lock.lock().unwrap();
match self.state.compare_exchange(EMPTY, PARKED, SeqCst, SeqCst) {
Ok(_) => {}
Err(NOTIFIED) => {
// We must read here, even though we know it will be `NOTIFIED`.
// This is because `unpark` may have been called again since we read
// `NOTIFIED` in the `compare_exchange` above. We must perform an
// acquire operation that synchronizes with that `unpark` to observe
// any writes it made before the call to unpark. To do that we must
// read from the write it made to `state`.
let old = self.state.swap(EMPTY, SeqCst);
assert_eq!(old, NOTIFIED, "park state changed unexpectedly");
return;
} // should consume this notification, so prohibit spurious wakeups in next park.
Err(_) => panic!("inconsistent park state"),
}
loop {
m = self.cvar.wait(m).unwrap();
match self.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst) {
Ok(_) => return, // got a notification
Err(_) => {} // spurious wakeup, go back to sleep
}
}
}
// This implementaiton doesn't require `unsafe`, but other implementations
// may assume this is only called by the thread that owns the Parker.
pub unsafe fn park_timeout(&self, dur: Duration) {
// Like `park` above we have a fast path for an already-notified thread, and
// afterwards we start coordinating for a sleep.
// return quickly.
if self.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst).is_ok() {
return;
}
let m = self.lock.lock().unwrap();
match self.state.compare_exchange(EMPTY, PARKED, SeqCst, SeqCst) {
Ok(_) => {}
Err(NOTIFIED) => {
// We must read again here, see `park`.
let old = self.state.swap(EMPTY, SeqCst);
assert_eq!(old, NOTIFIED, "park state changed unexpectedly");
return;
} // should consume this notification, so prohibit spurious wakeups in next park.
Err(_) => panic!("inconsistent park_timeout state"),
}
// Wait with a timeout, and if we spuriously wake up or otherwise wake up
// from a notification we just want to unconditionally set the state back to
// empty, either consuming a notification or un-flagging ourselves as
// parked.
let (_m, _result) = self.cvar.wait_timeout(m, dur).unwrap();
match self.state.swap(EMPTY, SeqCst) {
NOTIFIED => {} // got a notification, hurray!
PARKED => {} // no notification, alas
n => panic!("inconsistent park_timeout state: {}", n),
}
}
pub fn unpark(&self) {
// To ensure the unparked thread will observe any writes we made
// before this call, we must perform a release operation that `park`
// can synchronize with. To do that we must write `NOTIFIED` even if
// `state` is already `NOTIFIED`. That is why this must be a swap
// rather than a compare-and-swap that returns if it reads `NOTIFIED`
// on failure.
match self.state.swap(NOTIFIED, SeqCst) {
EMPTY => return, // no one was waiting
NOTIFIED => return, // already unparked
PARKED => {} // gotta go wake someone up
_ => panic!("inconsistent state in unpark"),
}
// There is a period between when the parked thread sets `state` to
// `PARKED` (or last checked `state` in the case of a spurious wake
// up) and when it actually waits on `cvar`. If we were to notify
// during this period it would be ignored and then when the parked
// thread went to sleep it would never wake up. Fortunately, it has
// `lock` locked at this stage so we can acquire `lock` to wait until
// it is ready to receive the notification.
//
// Releasing `lock` before the call to `notify_one` means that when the
// parked thread wakes it doesn't get woken only to have to wait for us
// to release `lock`.
drop(self.lock.lock().unwrap());
self.cvar.notify_one()
}
}

9
library/std/src/sys_common/thread_parker/mod.rs Normal file
View File

@ -0,0 +1,9 @@
cfg_if::cfg_if! {
if #[cfg(any(target_os = "linux", target_os = "android"))] {
mod futex;
pub use futex::Parker;
} else {
mod generic;
pub use generic::Parker;
}
}

128
library/std/src/thread/mod.rs
View File

@ -159,13 +159,12 @@ use crate::num::NonZeroU64;
use crate::panic;
use crate::panicking;
use crate::str;
use crate::sync::atomic::AtomicUsize;
use crate::sync::atomic::Ordering::SeqCst;
use crate::sync::{Arc, Condvar, Mutex};
use crate::sync::Arc;
use crate::sys::thread as imp;
use crate::sys_common::mutex;
use crate::sys_common::thread;
use crate::sys_common::thread_info;
use crate::sys_common::thread_parker::Parker;
use crate::sys_common::{AsInner, IntoInner};
use crate::time::Duration;
@ -667,6 +666,8 @@ pub fn current() -> Thread {
///
/// [`channel`]: crate::sync::mpsc
/// [`join`]: JoinHandle::join
/// [`Condvar`]: crate::sync::Condvar
/// [`Mutex`]: crate::sync::Mutex
#[stable(feature = "rust1", since = "1.0.0")]
pub fn yield_now() {
imp::Thread::yield_now()
@ -712,6 +713,8 @@ pub fn yield_now() {
/// panic!()
/// }
/// ```
///
/// [Mutex]: crate::sync::Mutex
#[inline]
#[stable(feature = "rust1", since = "1.0.0")]
pub fn panicking() -> bool {
@ -779,11 +782,6 @@ pub fn sleep(dur: Duration) {
imp::Thread::sleep(dur)
}
// constants for park/unpark
const EMPTY: usize = 0;
const PARKED: usize = 1;
const NOTIFIED: usize = 2;
/// Blocks unless or until the current thread's token is made available.
///
/// A call to `park` does not guarantee that the thread will remain parked
@ -870,45 +868,11 @@ const NOTIFIED: usize = 2;
///
/// [`unpark`]: Thread::unpark
/// [`thread::park_timeout`]: park_timeout
//
// The implementation currently uses the trivial strategy of a Mutex+Condvar
// with wakeup flag, which does not actually allow spurious wakeups. In the
// future, this will be implemented in a more efficient way, perhaps along the lines of
// http://cr.openjdk.java.net/~stefank/6989984.1/raw_files/new/src/os/linux/vm/os_linux.cpp
// or futuxes, and in either case may allow spurious wakeups.
#[stable(feature = "rust1", since = "1.0.0")]
pub fn park() {
let thread = current();
// If we were previously notified then we consume this notification and
// return quickly.
if thread.inner.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst).is_ok() {
return;
}
// Otherwise we need to coordinate going to sleep
let mut m = thread.inner.lock.lock().unwrap();
match thread.inner.state.compare_exchange(EMPTY, PARKED, SeqCst, SeqCst) {
Ok(_) => {}
Err(NOTIFIED) => {
// We must read here, even though we know it will be `NOTIFIED`.
// This is because `unpark` may have been called again since we read
// `NOTIFIED` in the `compare_exchange` above. We must perform an
// acquire operation that synchronizes with that `unpark` to observe
// any writes it made before the call to unpark. To do that we must
// read from the write it made to `state`.
let old = thread.inner.state.swap(EMPTY, SeqCst);
assert_eq!(old, NOTIFIED, "park state changed unexpectedly");
return;
} // should consume this notification, so prohibit spurious wakeups in next park.
Err(_) => panic!("inconsistent park state"),
}
loop {
m = thread.inner.cvar.wait(m).unwrap();
match thread.inner.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst) {
Ok(_) => return, // got a notification
Err(_) => {} // spurious wakeup, go back to sleep
}
// SAFETY: park_timeout is called on the parker owned by this thread.
unsafe {
current().inner.parker.park();
}
}
@ -970,35 +934,9 @@ pub fn park_timeout_ms(ms: u32) {
/// ```
#[stable(feature = "park_timeout", since = "1.4.0")]
pub fn park_timeout(dur: Duration) {
let thread = current();
// Like `park` above we have a fast path for an already-notified thread, and
// afterwards we start coordinating for a sleep.
// return quickly.
if thread.inner.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst).is_ok() {
return;
}
let m = thread.inner.lock.lock().unwrap();
match thread.inner.state.compare_exchange(EMPTY, PARKED, SeqCst, SeqCst) {
Ok(_) => {}
Err(NOTIFIED) => {
// We must read again here, see `park`.
let old = thread.inner.state.swap(EMPTY, SeqCst);
assert_eq!(old, NOTIFIED, "park state changed unexpectedly");
return;
} // should consume this notification, so prohibit spurious wakeups in next park.
Err(_) => panic!("inconsistent park_timeout state"),
}
// Wait with a timeout, and if we spuriously wake up or otherwise wake up
// from a notification we just want to unconditionally set the state back to
// empty, either consuming a notification or un-flagging ourselves as
// parked.
let (_m, _result) = thread.inner.cvar.wait_timeout(m, dur).unwrap();
match thread.inner.state.swap(EMPTY, SeqCst) {
NOTIFIED => {} // got a notification, hurray!
PARKED => {} // no notification, alas
n => panic!("inconsistent park_timeout state: {}", n),
// SAFETY: park_timeout is called on the parker owned by this thread.
unsafe {
current().inner.parker.park_timeout(dur);
}
}
@ -1077,11 +1015,7 @@ impl ThreadId {
struct Inner {
name: Option<CString>, // Guaranteed to be UTF-8
id: ThreadId,
// state for thread park/unpark
state: AtomicUsize,
lock: Mutex<()>,
cvar: Condvar,
parker: Parker,
}
#[derive(Clone)]
@ -1115,13 +1049,7 @@ impl Thread {
let cname =
name.map(|n| CString::new(n).expect("thread name may not contain interior null bytes"));
Thread {
inner: Arc::new(Inner {
name: cname,
id: ThreadId::new(),
state: AtomicUsize::new(EMPTY),
lock: Mutex::new(()),
cvar: Condvar::new(),
}),
inner: Arc::new(Inner { name: cname, id: ThreadId::new(), parker: Parker::new() }),
}
}
@ -1156,33 +1084,9 @@ impl Thread {
/// parked_thread.join().unwrap();
/// ```
#[stable(feature = "rust1", since = "1.0.0")]
#[inline]
pub fn unpark(&self) {
// To ensure the unparked thread will observe any writes we made
// before this call, we must perform a release operation that `park`
// can synchronize with. To do that we must write `NOTIFIED` even if
// `state` is already `NOTIFIED`. That is why this must be a swap
// rather than a compare-and-swap that returns if it reads `NOTIFIED`
// on failure.
match self.inner.state.swap(NOTIFIED, SeqCst) {
EMPTY => return, // no one was waiting
NOTIFIED => return, // already unparked
PARKED => {} // gotta go wake someone up
_ => panic!("inconsistent state in unpark"),
}
// There is a period between when the parked thread sets `state` to
// `PARKED` (or last checked `state` in the case of a spurious wake
// up) and when it actually waits on `cvar`. If we were to notify
// during this period it would be ignored and then when the parked
// thread went to sleep it would never wake up. Fortunately, it has
// `lock` locked at this stage so we can acquire `lock` to wait until
// it is ready to receive the notification.
//
// Releasing `lock` before the call to `notify_one` means that when the
// parked thread wakes it doesn't get woken only to have to wait for us
// to release `lock`.
drop(self.inner.lock.lock().unwrap());
self.inner.cvar.notify_one()
self.inner.parker.unpark();
}
/// Gets the thread's unique identifier.