I'm working with Notify from Tokio and I'm interested in how a waiter can "miss" a notification.

There's this line from the Tokio docs

The Notified future is not guaranteed to receive wakeups from calls to notify_one() if it has not yet been polled. See the documentation for Notified::enable() for more details.

So my understanding is this code should (or at least can) hang forever if "worker: notifying" is printed before "main: awaiting notification" because the Notified future is not polled via the await until after notify_one() is called.
Playground Link

use std::sync::Arc;
use tokio::sync::Notify;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let notify = Arc::new(Notify::new());
    let notified = notify.notified(); // not yet polled

    let worker_notify = notify.clone();

    tokio::spawn(async move {
        worker_notify.notify_one();
        println!("worker: notifying");
    });

    sleep(Duration::from_millis(1000)).await;
    println!("main: awaiting notification");
    notified.await;
    println!("main: received notification");
}

But it always exits.

Elsewhere in the docs it says

If notify_one() is called before notified().await , then the next call to notified().await will complete immediately, consuming the permit. Any subsequent calls to notified().await will wait for a new permit.

(emphasis mine) which seems to contradict what it says above.

The docs also say

The Notified future is guaranteed to receive wakeups from notify_waiters() as soon as it has been created, even if it has not yet been polled.

My understanding is that you need to either use notify_waiters()
Playground Link

use std::sync::Arc;
use tokio::sync::Notify;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let notify = Arc::new(Notify::new());
    let notified = notify.notified(); // not yet polled

    let worker_notify = notify.clone();

    tokio::spawn(async move {
        worker_notify.notify_waiters(); // changed this line
        println!("worker: notifying");
    });

    sleep(Duration::from_millis(1000)).await;
    println!("main: awaiting notification");
    notified.await;
    println!("main: received notification");
}

or use enable()
Per the docs

Adds this future to the list of futures that are ready to receive wakeups from calls to notify_one.
Polling the future also adds it to the list, so this method should only be used if you want to add the future to the list before the first call to poll.

Playground link

use std::sync::Arc;
use tokio::sync::Notify;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let notify = Arc::new(Notify::new());
    let notified = notify.notified(); // not yet polled
    
    // added these two lines
    tokio::pin!(notified);
    notified.as_mut().enable();
    
    let worker_notify = notify.clone();
    
    tokio::spawn(async move {
        worker_notify.notify_one(); 
        println!("worker: notifying");
    });

    sleep(Duration::from_millis(1000)).await;
    println!("main: awaiting notification");
    notified.await;
    println!("main: received notification");
}

The presence of enable() in the API and fact the docs contrast how notify_waiters() works with how notify_one() works would imply a notification can be missed. But the fact I can't reproduce that and the docs say "If notify_one() is called before notified().await , then the next call to notified().await will complete immediately" has me very confused.

If I have the pattern of creating a Notify and passing it to another thread that will notify this thread and I don't want to miss the notification, what is the correct safe way to ensure I won't miss the notification?

1 post - 1 participant

Read full topic