Структура Notified

Добавляет этот future в список future, готовых получать пробуждения от вызовов notify_one.

Опрос future также добавляет его в список, поэтому этот метод следует использовать только в том случае, если вы хотите добавить future в список до первого вызова poll. (Фактически этот метод эквивалентен вызову poll, за исключением того, что Waker не регистрируется.)

Это не влияет на уведомления, отправленные с помощью notify_waiters: они получаются, если возникают после создания Notified, независимо от того, вызывались ли enable или poll.

Метод возвращает true, если Notified готов. Это происходит в следующих случаях:

1. Метод notify_waiters был вызван между созданием Notified и вызовом этого метода.
2. Это первый вызов enable или poll для этого future, и Notify удерживал разрешение, полученное при предыдущем вызове notify_one. В этом случае вызов использует разрешение.
3. Ранее future был активирован или опрошен, а затем помечен как готовый либо в результате использования разрешения из Notify, либо вызовом notify_one или notify_waiters, который удалил его из списка future, готовых получать пробуждения.

Если этот метод возвращает true, любые последующие вызовы poll для того же future немедленно вернут Poll::Ready.

Примеры

Неограниченный канал «многие производители — многие потребители» (mpmc).

Вызов enable важен, поскольку в противном случае, если параллельно выполняются два вызова recv и два вызова send, может произойти следующее:

1. Оба вызова try_recv возвращают None.
2. Оба новых элемента добавляются в вектор.
3. Метод notify_one вызывается дважды, но в Notify добавляется только одно разрешение.
4. Оба вызова recv доходят до future Notified. Один из них использует разрешение, а второй засыпает навсегда.

Добавив future Notified в список вызовом enable до try_recv, вызовы notify_one на третьем шаге удалили бы future из списка и пометили их как уведомлённые, вместо того чтобы добавлять разрешение в Notify. Это гарантирует, что оба future будут разбужены.

use tokio::sync::Notify;

use std::collections::VecDeque;
use std::sync::Mutex;

struct Channel<T> {
    messages: Mutex<VecDeque<T>>,
    notify_on_sent: Notify,
}

impl<T> Channel<T> {
    pub fn send(&self, msg: T) {
        let mut locked_queue = self.messages.lock().unwrap();
        locked_queue.push_back(msg);
        drop(locked_queue);

        // Send a notification to one of the calls currently
        // waiting in a call to `recv`.
        self.notify_on_sent.notify_one();
    }

    pub fn try_recv(&self) -> Option<T> {
        let mut locked_queue = self.messages.lock().unwrap();
        locked_queue.pop_front()
    }

    pub async fn recv(&self) -> T {
        let future = self.notify_on_sent.notified();
        tokio::pin!(future);

        loop {
            // Make sure that no wakeup is lost if we get
            // `None` from `try_recv`.
            future.as_mut().enable();

            if let Some(msg) = self.try_recv() {
                return msg;
            }

            // Wait for a call to `notify_one`.
            //
            // This uses `.as_mut()` to avoid consuming the future,
            // which lets us call `Pin::set` below.
            future.as_mut().await;

            // Reset the future in case another call to
            // `try_recv` got the message before us.
            future.set(self.notify_on_sent.notified());
        }
    }
}

Возвращает аргумент без изменений.

Вызывает U::from(self).

То есть результат этого преобразования определяется реализацией From<T> for U.