What if we don't use channels?


Back to index

You may have noticed a problem when wrapping datastructures in asynchronous locking mechanisms

Insert items into collection on task 1 and busy-loop on task 2

// task 1
collection.lock().await.push_back(42);

// task2
let num = loop {
    if let Some(num) = collection.lock().await.pop_front() {
        num
    } else {
        smol::future::yield_now().await;
    }
};

yield_now


pub struct YieldNow(bool);

impl Future for YieldNow {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        if !self.0 {
            self.0 = true;
            cx.waker().wake_by_ref();
            Poll::Pending
        } else {
            Poll::Ready(())
        }
    }
}

Hot take 🔥 busy loops are bad

Solution? WakeNotify pattern

  • Wrap any collection in Notify<T>
  • Poll once, then store a waker
  • When inserting, call wake()
use core::task::Waker;
pub struct Notify<T> {
    inner: T,
    waker: Option<Waker>
}

WakeNotify pattern

// A nice and short type signature :)
let collection = Arc::new(Mutex::new(Notify::new(VecDeque::new())));

// task 1
let mut mg = c.lock().await;
mg.push_back(42);
Notify::wake(&*mg);
// mg goes out of scope to de-lock the Mutex

The poll task

futures::future::poll_fn(|ctx| {
    let mut lock = Box::pin(t.lock()); // We box this future to easily be able to pin it
    match Pin::new(&mut lock).poll(ctx) { // Then poll it for progress
        Poll::Ready(ref mut mg) => match mg.pop_front() {
            Some(v) => Poll::Ready(v), // Return data if there was any
            None => {                  // Otherwise install a Waker
                Notify::add_waker(mg, ctx.waker().clone());
                Poll::Pending
            }
        },
        _ => Poll::Pending, // If we were not able to acquire a lock,
                            // Mutex will wake us once we can!
    }
})
.await;

task-notify.png

https://crates.io/crates/task-notify

tokio-notify.png

Oh about those locks…

Async locks

Don't use std::sync types to lock!

  • tokio::sync::{Mutex, RwLock, Barrier}
  • async_std::sync::{Mutex, RwLock, Barrier}
  • async_lock::{Mutex, RwLock, Barrior}

Back to index