结构体 Notified

Source

pub struct Notified<'a> { /* private fields */ }

展开描述

由 Notify::notified() 返回的 future。

此 future 是 fused 的，因此一旦它完成，任何对 poll 的后续调用将立即返回 Poll::Ready。

实现§

Source §

impl Notified<'_>

Source

pub fn enable(self: Pin<&mut Self>) -> bool

将此 future 添加到可以从 notify_one 接收唤醒的 future 列表中。

poll 该 future 也会将其添加到列表中，因此仅当您想在第一次调用 poll 之前将该 future 添加到列表中时才应使用此方法。（实际上，此方法等价于调用 poll，只是不注册 Waker）。

这对使用 notify_waiters 发送的通知没有影响，只要它们在 Notified 创建后发生，无论是否已调用 enable 或 poll 都会被接收。

如果 Notified 已就绪，此方法返回 true。发生这种情况的场景如下：

The notify_waiters method was called between the creation of the Notified and the call to this method.
This is the first call to enable or poll on this future, and the Notify was holding a permit from a previous call to notify_one. The call consumes the permit in that case.
The future has previously been enabled or polled, and it has since then been marked ready by either consuming a permit from the Notify, or by a call to notify_one or notify_waiters that removed it from the list of futures ready to receive wakeups.

如果此方法返回 true，那么对同一 future 的任何后续 poll 调用都将立即返回 Poll::Ready。

§示例

无界多生产者多消费者 (mpmc) 通道。

调用 enable 很重要，因为否则如果并行有两个 recv 调用和两个 send 调用，可能会发生以下情况：

Both calls to try_recv return None.
Both new elements are added to the vector.
The notify_one method is called twice, adding only a single permit to the Notify.
Both calls to recv reach the Notified future. One of them consumes the permit, and the other sleeps forever.

通过在 try_recv 之前调用 enable 将 Notified future 添加到列表中，第三步中的 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());
        }
    }
}