Skip to content
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.

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

Reimplement std's thread parker on top of events on SGX #98391

Merged
merged 3 commits into from
Dec 10, 2022
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
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
1 change: 1 addition & 0 deletions library/std/src/sys/sgx/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@ pub mod process;
pub mod stdio;
pub mod thread;
pub mod thread_local_key;
pub mod thread_parker;
pub mod time;

mod condvar;
Expand Down
21 changes: 9 additions & 12 deletions library/std/src/sys/sgx/thread.rs
Original file line number Diff line number Diff line change
Expand Up @@ -65,39 +65,36 @@ mod task_queue {
/// execution. The signal is sent once all TLS destructors have finished at
/// which point no new thread locals should be created.
pub mod wait_notify {
use super::super::waitqueue::{SpinMutex, WaitQueue, WaitVariable};
use super::super::thread_parker::Parker;
use crate::pin::Pin;
use crate::sync::Arc;

pub struct Notifier(Arc<SpinMutex<WaitVariable<bool>>>);
pub struct Notifier(Arc<Parker>);

impl Notifier {
/// Notify the waiter. The waiter is either notified right away (if
/// currently blocked in `Waiter::wait()`) or later when it calls the
/// `Waiter::wait()` method.
pub fn notify(self) {
let mut guard = self.0.lock();
*guard.lock_var_mut() = true;
let _ = WaitQueue::notify_one(guard);
Pin::new(&*self.0).unpark()
}
}

pub struct Waiter(Arc<SpinMutex<WaitVariable<bool>>>);
pub struct Waiter(Arc<Parker>);

impl Waiter {
/// Wait for a notification. If `Notifier::notify()` has already been
/// called, this will return immediately, otherwise the current thread
/// is blocked until notified.
pub fn wait(self) {
let guard = self.0.lock();
if *guard.lock_var() {
return;
}
WaitQueue::wait(guard, || {});
// This is not actually `unsafe`, but it uses the `Parker` API,
// which needs `unsafe` on some platforms.
unsafe { Pin::new(&*self.0).park() }
}
}

pub fn new() -> (Notifier, Waiter) {
let inner = Arc::new(SpinMutex::new(WaitVariable::new(false)));
let inner = Arc::new(Parker::new_internal());
(Notifier(inner.clone()), Waiter(inner))
}
}
Expand Down
93 changes: 93 additions & 0 deletions library/std/src/sys/sgx/thread_parker.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,93 @@
//! Thread parking based on SGX events.

use super::abi::{thread, usercalls};
use crate::io::ErrorKind;
use crate::pin::Pin;
use crate::ptr::{self, NonNull};
use crate::sync::atomic::AtomicPtr;
use crate::sync::atomic::Ordering::{Acquire, Relaxed, Release};
use crate::time::Duration;
use fortanix_sgx_abi::{EV_UNPARK, WAIT_INDEFINITE};

const EMPTY: *mut u8 = ptr::invalid_mut(0);
joboet marked this conversation as resolved.
Show resolved Hide resolved
/// The TCS structure must be page-aligned, so this cannot be a valid pointer
const NOTIFIED: *mut u8 = ptr::invalid_mut(1);

pub struct Parker {
state: AtomicPtr<u8>,
}

impl Parker {
/// Construct the thread parker. The UNIX parker implementation
/// requires this to happen in-place.
pub unsafe fn new(parker: *mut Parker) {
unsafe { parker.write(Parker::new_internal()) }
}

pub(super) fn new_internal() -> Parker {
Parker { state: AtomicPtr::new(EMPTY) }
}

// This implementation doesn't require `unsafe` and `Pin`, but other implementations do.
pub unsafe fn park(self: Pin<&Self>) {
joboet marked this conversation as resolved.
Show resolved Hide resolved
let tcs = thread::current().as_ptr();

if self.state.load(Acquire) != NOTIFIED {
joboet marked this conversation as resolved.
Show resolved Hide resolved
if self.state.compare_exchange(EMPTY, tcs, Acquire, Acquire).is_ok() {
// Loop to guard against spurious wakeups.
loop {
let event = usercalls::wait(EV_UNPARK, WAIT_INDEFINITE).unwrap();
assert!(event & EV_UNPARK == EV_UNPARK);
if self.state.load(Acquire) == NOTIFIED {
break;
}
}
}
}

// At this point, the token was definately read with acquire ordering,
// so this can be a store.
self.state.store(EMPTY, Relaxed);
}

// This implementation doesn't require `unsafe` and `Pin`, but other implementations do.
pub unsafe fn park_timeout(self: Pin<&Self>, dur: Duration) {
let timeout = u128::min(dur.as_nanos(), WAIT_INDEFINITE as u128 - 1) as u64;
let tcs = thread::current().as_ptr();

if self.state.load(Acquire) != NOTIFIED {
if self.state.compare_exchange(EMPTY, tcs, Acquire, Acquire).is_ok() {
match usercalls::wait(EV_UNPARK, timeout) {
Ok(event) => assert!(event & EV_UNPARK == EV_UNPARK),
Err(e) => {
assert!(matches!(e.kind(), ErrorKind::TimedOut | ErrorKind::WouldBlock))
}
}

// Swap to provide acquire ordering even if the timeout occurred
// before the token was set. This situation can result in spurious
// wakeups on the next call to `park_timeout`, but it is better to let
// those be handled by the user than do some perhaps unnecessary, but
// always expensive guarding.
self.state.swap(EMPTY, Acquire);
return;
}
}

// The token was already read with `acquire` ordering, this can be a store.
self.state.store(EMPTY, Relaxed);
}

// This implementation doesn't require `Pin`, but other implementations do.
pub fn unpark(self: Pin<&Self>) {
joboet marked this conversation as resolved.
Show resolved Hide resolved
let state = self.state.swap(NOTIFIED, Release);

if !matches!(state, EMPTY | NOTIFIED) {
// There is a thread waiting, wake it up.
let tcs = NonNull::new(state).unwrap();
// This will fail if the thread has already terminated by the time the signal is send,
// but that is OK.
let _ = usercalls::send(EV_UNPARK, Some(tcs));
}
}
}
2 changes: 2 additions & 0 deletions library/std/src/sys_common/thread_parker/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -13,6 +13,8 @@ cfg_if::cfg_if! {
pub use crate::sys::thread_parker::Parker;
} else if #[cfg(target_family = "unix")] {
pub use crate::sys::thread_parker::Parker;
} else if #[cfg(all(target_vendor = "fortanix", target_env = "sgx"))] {
pub use crate::sys::thread_parker::Parker;
} else {
mod generic;
pub use generic::Parker;
Expand Down