# Thread Looping
Loop in a thread effectively, like giving the job some period of time that allows us return
operability to the OS.
```rust
{
/// Moves the ticker into woken state.
fn wake(&mut self) {
if self.sleeping != 0 {
let mut sleepers = self.state.sleepers.lock().unwrap();
sleepers.remove(self.sleeping);
self.state
.notified
.store(sleepers.is_notified(), Ordering::Release);
}
self.sleeping = 0;
}
/// Waits for the next runnable task to run.
async fn runnable(&mut self) -> Runnable {
self.runnable_with(|| self.state.queue.pop().ok()).await
}
/// Waits for the next runnable task to run, given a function that searches for a task.
async fn runnable_with(&mut self, mut search: impl FnMut() -> Option<Runnable>) -> Runnable {
future::poll_fn(|cx| {
loop {
match search() {
None => {
// Move to sleeping and unnotified state.
if !self.sleep(cx.waker()) {
// If already sleeping and unnotified, return.
return Poll::Pending;
}
}
Some(r) => {
// Wake up.
self.wake();
// Notify another ticker now to pick up where this ticker left off, just in
// case running the task takes a long time.
self.state.notify();
return Poll::Ready(r);
}
}
}
})
.await
}
}
impl Drop for Ticker<'_> {
fn drop(&mut self) {
// If this ticker is in sleeping state, it must be removed from the sleepers list.
if self.sleeping != 0 {
let mut sleepers = self.state.sleepers.lock().unwrap();
let notified = sleepers.remove(self.sleeping);
self.state
.notified
.store(sleepers.is_notified(), Ordering::Release);
// If this ticker was notified, then notify another ticker.
if notified {
drop(sleepers);
self.state.notify();
}
}
}
}
/// Creates state for a new executor.
const fn new() -> State {
State {
queue: ConcurrentQueue::unbounded(),
local_queues: RwLock::new(Vec::new()),
notified: AtomicBool::new(true),
sleepers: Mutex::new(Sleepers {
count: 0,
wakers: Vec::new(),
free_ids: Vec::new(),
}),
active: Mutex::new(Slab::new()),
}
}
/// Returns a reference to currently active tasks.
fn active(&self) -> MutexGuard<'_, Slab<Waker>> {
self.active.lock().unwrap_or_else(|e| e.into_inner())
}
/// Notifies a sleeping ticker.
#[inline]
fn notify(&self) {
if self
.notified
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
.is_ok()
{
let waker = self.sleepers.lock().unwrap().notify();
if let Some(w) = waker {
w.wake();
}
}
}
pub(crate) fn try_tick(&self) -> bool {
match self.queue.pop() {
Err(_) => false,
Ok(runnable) => {
// Notify another ticker now to pick up where this ticker left off, just in case
// running the task takes a long time.
self.notify();
// Run the task.
runnable.run();
true
}
}
}
pub(crate) async fn tick(&self) {
let runnable = Ticker::new(self).runnable().await;
runnable.run();
}
pub async fn run<T>(&self, future: impl Future<Output = T>) -> T {
let mut runner = Runner::new(self);
let mut rng = fastrand::Rng::new();
// A future that runs tasks forever.
let run_forever = async {
loop {
for _ in 0..200 {
let runnable = runner.runnable(&mut rng).await;
runnable.run();
}
future::yield_now().await;
}
};
// Run `future` and `run_forever` concurrently until `future` completes.
future.or(run_forever).await
}
}
```
Block on
```rust
use std::cell::{Cell, RefCell};
use std::future::Future;
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
use std::sync::Arc;
use std::task::Waker;
use std::task::{Context, Poll};
use std::thread;
use std::time::{Duration, Instant};
use async_lock::OnceCell;
use futures_lite::pin;
use parking::Parker;
use crate::reactor::Reactor;
/// Number of currently active `block_on()` invocations.
static BLOCK_ON_COUNT: AtomicUsize = AtomicUsize::new(0);
/// Unparker for the "async-io" thread.
fn unparker() -> &'static parking::Unparker {
static UNPARKER: OnceCell<parking::Unparker> = OnceCell::new();
UNPARKER.get_or_init_blocking(|| {
let (parker, unparker) = parking::pair();
// Spawn a helper thread driving the reactor.
//
// Note that this thread is not exactly necessary, it's only here to help push things
// forward if there are no `Parker`s around or if `Parker`s are just idling and never
// parking.
thread::Builder::new()
.name("async-io".to_string())
.spawn(move || main_loop(parker))
.expect("cannot spawn async-io thread");
unparker
})
}
/// Initializes the "async-io" thread.
pub(crate) fn init() {
let _ = unparker();
}
/// The main loop for the "async-io" thread.
fn main_loop(parker: parking::Parker) {
let span = tracing::trace_span!("async_io::main_loop");
let _enter = span.enter();
// The last observed reactor tick.
let mut last_tick = 0;
// Number of sleeps since this thread has called `react()`.
let mut sleeps = 0u64;
loop {
let tick = Reactor::get().ticker();
if last_tick == tick {
let reactor_lock = if sleeps >= 10 {
// If no new ticks have occurred for a while, stop sleeping and spinning in
// this loop and just block on the reactor lock.
Some(Reactor::get().lock())
} else {
Reactor::get().try_lock()
};
if let Some(mut reactor_lock) = reactor_lock {
tracing::trace!("waiting on I/O");
reactor_lock.react(None).ok();
last_tick = Reactor::get().ticker();
sleeps = 0;
}
} else {
last_tick = tick;
}
if BLOCK_ON_COUNT.load(Ordering::SeqCst) > 0 {
// Exponential backoff from 50us to 10ms.
let delay_us = [50, 75, 100, 250, 500, 750, 1000, 2500, 5000]
.get(sleeps as usize)
.unwrap_or(&10_000);
tracing::trace!("sleeping for {} us", delay_us);
if parker.park_timeout(Duration::from_micros(*delay_us)) {
tracing::trace!("notified");
// If notified before timeout, reset the last tick and the sleep counter.
last_tick = Reactor::get().ticker();
sleeps = 0;
} else {
sleeps += 1;
}
}
}
}
/// Blocks the current thread on a future, processing I/O events when idle.
///
/// # Examples
///
/// ```
/// use async_io::Timer;
/// use std::time::Duration;
///
/// async_io::block_on(async {
/// // This timer will likely be processed by the current
/// // thread rather than the fallback "async-io" thread.
/// Timer::after(Duration::from_millis(1)).await;
/// });
/// ```
pub fn block_on<T>(future: impl Future<Output = T>) -> T {
let span = tracing::trace_span!("async_io::block_on");
let _enter = span.enter();
// Increment `BLOCK_ON_COUNT` so that the "async-io" thread becomes less aggressive.
BLOCK_ON_COUNT.fetch_add(1, Ordering::SeqCst);
// Make sure to decrement `BLOCK_ON_COUNT` at the end and wake the "async-io" thread.
let _guard = CallOnDrop(|| {
BLOCK_ON_COUNT.fetch_sub(1, Ordering::SeqCst);
unparker().unpark();
});
// Creates a parker and an associated waker that unparks it.
fn parker_and_waker() -> (Parker, Waker, Arc<AtomicBool>) {
// Parker and unparker for notifying the current thread.
let (p, u) = parking::pair();
// This boolean is set to `true` when the current thread is blocked on I/O.
let io_blocked = Arc::new(AtomicBool::new(false));
// Prepare the waker.
let waker = BlockOnWaker::create(io_blocked.clone(), u);
(p, waker, io_blocked)
}
thread_local! {
// Cached parker and waker for efficiency.
static CACHE: RefCell<(Parker, Waker, Arc<AtomicBool>)> = RefCell::new(parker_and_waker());
// Indicates that the current thread is polling I/O, but not necessarily blocked on it.
static IO_POLLING: Cell<bool> = const { Cell::new(false) };
}
struct BlockOnWaker {
io_blocked: Arc<AtomicBool>,
unparker: parking::Unparker,
}
impl BlockOnWaker {
fn create(io_blocked: Arc<AtomicBool>, unparker: parking::Unparker) -> Waker {
Waker::from(Arc::new(BlockOnWaker {
io_blocked,
unparker,
}))
}
}
impl std::task::Wake for BlockOnWaker {
fn wake_by_ref(self: &Arc<Self>) {
if self.unparker.unpark() {
// Check if waking from another thread and if currently blocked on I/O.
if !IO_POLLING.with(Cell::get) && self.io_blocked.load(Ordering::SeqCst) {
Reactor::get().notify();
}
}
}
fn wake(self: Arc<Self>) {
self.wake_by_ref()
}
}
CACHE.with(|cache| {
// Try grabbing the cached parker and waker.
let tmp_cached;
let tmp_fresh;
let (p, waker, io_blocked) = match cache.try_borrow_mut() {
Ok(cache) => {
// Use the cached parker and waker.
tmp_cached = cache;
&*tmp_cached
}
Err(_) => {
// Looks like this is a recursive `block_on()` call.
// Create a fresh parker and waker.
tmp_fresh = parker_and_waker();
&tmp_fresh
}
};
pin!(future);
let cx = &mut Context::from_waker(waker);
loop {
// Poll the future.
if let Poll::Ready(t) = future.as_mut().poll(cx) {
// Ensure the cached parker is reset to the unnotified state for future block_on calls,
// in case this future called wake and then immediately returned Poll::Ready.
p.park_timeout(Duration::from_secs(0));
tracing::trace!("completed");
return t;
}
// Check if a notification was received.
if p.park_timeout(Duration::from_secs(0)) {
tracing::trace!("notified");
// Try grabbing a lock on the reactor to process I/O events.
if let Some(mut reactor_lock) = Reactor::get().try_lock() {
// First let wakers know this parker is processing I/O events.
IO_POLLING.with(|io| io.set(true));
let _guard = CallOnDrop(|| {
IO_POLLING.with(|io| io.set(false));
});
// Process available I/O events.
reactor_lock.react(Some(Duration::from_secs(0))).ok();
}
continue;
}
// Try grabbing a lock on the reactor to wait on I/O.
if let Some(mut reactor_lock) = Reactor::get().try_lock() {
// Record the instant at which the lock was grabbed.
let start = Instant::now();
loop {
// First let wakers know this parker is blocked on I/O.
IO_POLLING.with(|io| io.set(true));
io_blocked.store(true, Ordering::SeqCst);
let _guard = CallOnDrop(|| {
IO_POLLING.with(|io| io.set(false));
io_blocked.store(false, Ordering::SeqCst);
});
// Check if a notification has been received before `io_blocked` was updated
// because in that case the reactor won't receive a wakeup.
if p.park_timeout(Duration::from_secs(0)) {
tracing::trace!("notified");
break;
}
// Wait for I/O events.
tracing::trace!("waiting on I/O");
reactor_lock.react(None).ok();
// Check if a notification has been received.
if p.park_timeout(Duration::from_secs(0)) {
tracing::trace!("notified");
break;
}
// Check if this thread been handling I/O events for a long time.
if start.elapsed() > Duration::from_micros(500) {
tracing::trace!("stops hogging the reactor");
// This thread is clearly processing I/O events for some other threads
// because it didn't get a notification yet. It's best to stop hogging the
// reactor and give other threads a chance to process I/O events for
// themselves.
drop(reactor_lock);
// Unpark the "async-io" thread in case no other thread is ready to start
// processing I/O events. This way we prevent a potential latency spike.
unparker().unpark();
// Wait for a notification.
p.park();
break;
}
}
} else {
// Wait for an actual notification.
tracing::trace!("sleep until notification");
p.park();
}
}
})
}
/// Runs a closure when dropped.
struct CallOnDrop<F: Fn()>(F);
impl<F: Fn()> Drop for CallOnDrop<F> {
fn drop(&mut self) {
(self.0)();
}
}
```