mirror of https://github.com/stjepang/smol
work
This commit is contained in:
parent
e6bd9229f7
commit
0408c19531
|
@ -0,0 +1,187 @@
|
|||
use std::fmt;
|
||||
use std::marker::PhantomData;
|
||||
use std::sync::atomic::AtomicUsize;
|
||||
use std::sync::atomic::Ordering::SeqCst;
|
||||
use std::sync::{Arc, Condvar, Mutex};
|
||||
use std::time::{Duration, Instant};
|
||||
|
||||
use crate::reactor::{Reactor, ReactorLock};
|
||||
|
||||
pub(crate) struct IoParker {
|
||||
unparker: IoUnparker,
|
||||
_marker: PhantomData<*const ()>,
|
||||
}
|
||||
|
||||
unsafe impl Send for IoParker {}
|
||||
|
||||
impl IoParker {
|
||||
pub fn new() -> IoParker {
|
||||
IoParker {
|
||||
unparker: IoUnparker {
|
||||
inner: Arc::new(Inner {
|
||||
state: AtomicUsize::new(EMPTY),
|
||||
lock: Mutex::new(()),
|
||||
cvar: Condvar::new(),
|
||||
}),
|
||||
},
|
||||
_marker: PhantomData,
|
||||
}
|
||||
}
|
||||
|
||||
pub fn park(&self) {
|
||||
self.unparker.inner.park(None);
|
||||
}
|
||||
|
||||
pub fn park_timeout(&self, timeout: Duration) -> bool {
|
||||
self.unparker.inner.park(Some(timeout))
|
||||
}
|
||||
|
||||
pub fn park_deadline(&self, deadline: Instant) -> bool {
|
||||
self.unparker
|
||||
.inner
|
||||
.park(Some(deadline.saturating_duration_since(Instant::now())))
|
||||
}
|
||||
|
||||
pub fn unpark(&self) {
|
||||
self.unparker.unpark()
|
||||
}
|
||||
|
||||
pub fn unparker(&self) -> IoUnparker {
|
||||
self.unparker.clone()
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Debug for IoParker {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.pad("IoParker { .. }")
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct IoUnparker {
|
||||
inner: Arc<Inner>,
|
||||
}
|
||||
|
||||
unsafe impl Send for IoUnparker {}
|
||||
unsafe impl Sync for IoUnparker {}
|
||||
|
||||
impl IoUnparker {
|
||||
pub fn unpark(&self) {
|
||||
self.inner.unpark()
|
||||
}
|
||||
}
|
||||
|
||||
impl fmt::Debug for IoUnparker {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
f.pad("IoUnparker { .. }")
|
||||
}
|
||||
}
|
||||
|
||||
impl Clone for IoUnparker {
|
||||
fn clone(&self) -> IoUnparker {
|
||||
IoUnparker {
|
||||
inner: self.inner.clone(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
const EMPTY: usize = 0;
|
||||
const PARKED: usize = 1;
|
||||
const NOTIFIED: usize = 2;
|
||||
|
||||
struct Inner {
|
||||
state: AtomicUsize,
|
||||
lock: Mutex<()>,
|
||||
cvar: Condvar,
|
||||
}
|
||||
|
||||
impl Inner {
|
||||
fn park(&self, timeout: Option<Duration>) -> bool {
|
||||
// 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 true;
|
||||
}
|
||||
|
||||
// If the timeout is zero, then there is no need to actually block.
|
||||
if let Some(dur) = timeout {
|
||||
if dur == Duration::from_millis(0) {
|
||||
if let Some(mut reactor_lock) = Reactor::get().try_lock() {
|
||||
reactor_lock.poll().expect("failure while polling I/O");
|
||||
}
|
||||
return false;
|
||||
}
|
||||
}
|
||||
|
||||
// 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(_) => {}
|
||||
// Consume this notification to avoid spurious wakeups in the next park.
|
||||
Err(NOTIFIED) => {
|
||||
// We must read `state` 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 true;
|
||||
}
|
||||
Err(n) => panic!("inconsistent park_timeout state: {}", n),
|
||||
}
|
||||
|
||||
match timeout {
|
||||
None => {
|
||||
loop {
|
||||
// Block the current thread on the conditional variable.
|
||||
m = self.cvar.wait(m).unwrap();
|
||||
|
||||
match self.state.compare_exchange(NOTIFIED, EMPTY, SeqCst, SeqCst) {
|
||||
Ok(_) => return true, // got a notification
|
||||
Err(_) => {} // spurious wakeup, go back to sleep
|
||||
}
|
||||
}
|
||||
}
|
||||
Some(timeout) => {
|
||||
// Wait with a timeout, and if we spuriously wake up or otherwise wake up from a
|
||||
// notification we just want to unconditionally set `state` back to `EMPTY`, either
|
||||
// consuming a notification or un-flagging ourselves as parked.
|
||||
let (_m, _result) = self.cvar.wait_timeout(m, timeout).unwrap();
|
||||
|
||||
match self.state.swap(EMPTY, SeqCst) {
|
||||
NOTIFIED => true, // got a notification
|
||||
PARKED => false, // no notification
|
||||
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 wakeup) 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();
|
||||
}
|
||||
}
|
|
@ -120,6 +120,7 @@ mod block_on;
|
|||
mod blocking;
|
||||
mod context;
|
||||
mod io_event;
|
||||
mod io_parking;
|
||||
mod reactor;
|
||||
mod run;
|
||||
mod sys;
|
||||
|
|
69
src/run.rs
69
src/run.rs
|
@ -14,7 +14,7 @@ use crate::io_event::IoEvent;
|
|||
use crate::reactor::{Reactor, ReactorLock};
|
||||
use crate::thread_local::ThreadLocalExecutor;
|
||||
use crate::throttle;
|
||||
use crate::work_stealing::WorkStealingExecutor;
|
||||
use crate::work_stealing::{WorkStealingExecutor, Worker};
|
||||
|
||||
/// Runs executors and polls the reactor.
|
||||
///
|
||||
|
@ -95,30 +95,22 @@ use crate::work_stealing::WorkStealingExecutor;
|
|||
/// }
|
||||
/// ```
|
||||
pub fn run<T>(future: impl Future<Output = T>) -> T {
|
||||
// Create a thread-local executor and a worker in the work-stealing executor.
|
||||
let local = ThreadLocalExecutor::new();
|
||||
let event = IoEvent::new().expect("cannot create an `IoEvent`");
|
||||
let ws_executor = WorkStealingExecutor::get();
|
||||
let worker = ws_executor.worker();
|
||||
let reactor = Reactor::get();
|
||||
|
||||
// Create a waker that triggers an I/O event in the thread-local scheduler.
|
||||
let ev = local.event().clone();
|
||||
let ev = event.clone();
|
||||
let waker = async_task::waker_fn(move || ev.notify());
|
||||
let cx = &mut Context::from_waker(&waker);
|
||||
futures_util::pin_mut!(future);
|
||||
|
||||
// Set up tokio (if enabled) and the thread-locals before execution begins.
|
||||
let enter = context::enter;
|
||||
let enter = |f| local.enter(|| enter(f));
|
||||
let enter = |f| worker.enter(|| enter(f));
|
||||
|
||||
enter(|| {
|
||||
// A list of I/O events that indicate there is work to do.
|
||||
let io_events = [local.event(), ws_executor.event()];
|
||||
|
||||
// Number of times this thread has yielded because it didn't find any work.
|
||||
let mut yields = 0;
|
||||
|
||||
// We run four components at the same time, treating them all fairly and making sure none
|
||||
// of them get starved:
|
||||
//
|
||||
|
@ -140,39 +132,25 @@ pub fn run<T>(future: impl Future<Output = T>) -> T {
|
|||
// This way we make sure that if any changes happen that might give us new work will
|
||||
// unblock epoll/kevent/wepoll and let us continue the loop.
|
||||
loop {
|
||||
// 1. Poll the main future.
|
||||
if let Poll::Ready(val) = throttle::setup(|| future.as_mut().poll(cx)) {
|
||||
return val;
|
||||
}
|
||||
|
||||
// 2. Run a batch of tasks in the thread-local executor.
|
||||
let more_local = local.execute();
|
||||
// 3. Run a batch of tasks in the work-stealing executor.
|
||||
let more_worker = worker.execute();
|
||||
let more = worker.execute();
|
||||
|
||||
if more {
|
||||
if let Some(mut reactor_lock) = reactor.try_lock() {
|
||||
reactor_lock.poll().expect("failure while polling I/O");
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
// 4. Poll the reactor.
|
||||
if let Some(reactor_lock) = reactor.try_lock() {
|
||||
yields = 0;
|
||||
react(reactor_lock, &io_events, more_local || more_worker);
|
||||
react(&worker, reactor_lock, &event);
|
||||
continue;
|
||||
}
|
||||
|
||||
// If there is more work in the thread-local or the work-stealing executor, continue.
|
||||
if more_local || more_worker {
|
||||
yields = 0;
|
||||
continue;
|
||||
}
|
||||
|
||||
// Yield a few times if no work is found.
|
||||
yields += 1;
|
||||
if yields <= 2 {
|
||||
thread::yield_now();
|
||||
continue;
|
||||
}
|
||||
|
||||
// If still no work is found, stop yielding and block the thread.
|
||||
yields = 0;
|
||||
|
||||
// Prepare for blocking until the reactor is locked or `local.event()` is triggered.
|
||||
//
|
||||
// Note that there is no need to wait for `ws_executor.event()`. If we lock the reactor
|
||||
|
@ -183,16 +161,16 @@ pub fn run<T>(future: impl Future<Output = T>) -> T {
|
|||
// and will unlock it if there is more work to do because every worker triggers the I/O
|
||||
// event whenever it finds a runnable task.
|
||||
let lock = reactor.lock();
|
||||
let notified = local.event().notified();
|
||||
let notified = event.notified();
|
||||
futures_util::pin_mut!(lock);
|
||||
futures_util::pin_mut!(notified);
|
||||
|
||||
// Block until either the reactor is locked or `local.event()` is triggered.
|
||||
if let Either::Left((reactor_lock, _)) = block_on(future::select(lock, notified)) {
|
||||
react(reactor_lock, &io_events, false);
|
||||
react(&worker, reactor_lock, &event);
|
||||
} else {
|
||||
// Clear `local.event()` because it was triggered.
|
||||
local.event().clear();
|
||||
event.clear();
|
||||
}
|
||||
}
|
||||
})
|
||||
|
@ -204,25 +182,14 @@ pub fn run<T>(future: impl Future<Output = T>) -> T {
|
|||
/// Otherwise, the current thread waits on it until a timer fires or an I/O event occurs.
|
||||
///
|
||||
/// I/O events are cleared at the end of this function.
|
||||
fn react(mut reactor_lock: ReactorLock<'_>, io_events: &[&IoEvent], mut more_tasks: bool) {
|
||||
// Clear all I/O events and check if any of them were triggered.
|
||||
for ev in io_events {
|
||||
if ev.clear() {
|
||||
more_tasks = true;
|
||||
}
|
||||
}
|
||||
|
||||
if more_tasks {
|
||||
fn react(worker: &Worker<'_>, mut reactor_lock: ReactorLock<'_>, event: &IoEvent) {
|
||||
if event.clear() {
|
||||
// If there might be more tasks to run, just poll without blocking.
|
||||
reactor_lock.poll().expect("failure while polling I/O");
|
||||
} else {
|
||||
// Otherwise, block until the first I/O event or a timer.
|
||||
reactor_lock.wait().expect("failure while waiting on I/O");
|
||||
|
||||
// Clear all I/O events before dropping the lock. This is not really necessary, but
|
||||
// clearing flags here might prevent a redundant wakeup in the future.
|
||||
for ev in io_events {
|
||||
ev.clear();
|
||||
}
|
||||
event.clear();
|
||||
}
|
||||
}
|
||||
|
|
|
@ -21,7 +21,8 @@
|
|||
use std::cell::Cell;
|
||||
use std::future::Future;
|
||||
use std::panic;
|
||||
use std::sync::{Arc, RwLock};
|
||||
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
||||
use std::sync::{Arc, Mutex, RwLock};
|
||||
|
||||
use concurrent_queue::ConcurrentQueue;
|
||||
use once_cell::sync::Lazy;
|
||||
|
@ -53,6 +54,10 @@ pub(crate) struct WorkStealingExecutor {
|
|||
|
||||
/// An I/O event that is triggered whenever there might be available tasks to run.
|
||||
event: IoEvent,
|
||||
|
||||
notified: AtomicBool,
|
||||
|
||||
sleeping: Mutex<Vec<Arc<dyn Fn() + Send + Sync>>>,
|
||||
}
|
||||
|
||||
impl WorkStealingExecutor {
|
||||
|
@ -62,13 +67,20 @@ impl WorkStealingExecutor {
|
|||
injector: ConcurrentQueue::unbounded(),
|
||||
stealers: RwLock::new(Slab::new()),
|
||||
event: IoEvent::new().expect("cannot create an `IoEvent`"),
|
||||
notified: AtomicBool::new(false),
|
||||
sleeping: Mutex::new(Vec::new()),
|
||||
});
|
||||
&EXECUTOR
|
||||
}
|
||||
|
||||
/// Returns the event indicating there is a scheduled task.
|
||||
pub fn event(&self) -> &IoEvent {
|
||||
&self.event
|
||||
fn notify(&self) {
|
||||
// if !self.notified.compare_and_swap(false, true, Ordering::SeqCst) {
|
||||
// let mut sleeping = self.sleeping.lock().unwrap();
|
||||
// if let Some(callback) = sleeping.pop() {
|
||||
// callback();
|
||||
// }
|
||||
// }
|
||||
self.event.notify();
|
||||
}
|
||||
|
||||
/// Spawns a future onto this executor.
|
||||
|
@ -88,7 +100,7 @@ impl WorkStealingExecutor {
|
|||
self.injector.push(runnable).unwrap();
|
||||
|
||||
// Notify workers that there is a task in the injector queue.
|
||||
self.event.notify();
|
||||
self.notify();
|
||||
}
|
||||
};
|
||||
|
||||
|
@ -169,7 +181,7 @@ impl Worker<'_> {
|
|||
// need to worry about `search()` re-shuffling tasks between queues, which
|
||||
// races with other workers searching for tasks. Other workers might not
|
||||
// find a task while there is one! Notifying here avoids this problem.
|
||||
self.executor.event.notify();
|
||||
self.executor.notify();
|
||||
|
||||
// Run the task.
|
||||
if throttle::setup(|| r.run()) {
|
||||
|
@ -325,6 +337,6 @@ impl Drop for Worker<'_> {
|
|||
|
||||
// This task will not search for tasks anymore and therefore won't notify other workers if
|
||||
// new tasks are found. Notify another worker to start searching right away.
|
||||
self.executor.event.notify();
|
||||
self.executor.notify();
|
||||
}
|
||||
}
|
||||
|
|
Loading…
Reference in New Issue