结构体 Notified

源代码

pub struct Notified<'a> { /* 私有字段 */ }

展开描述

从 Notify::notified() 返回的 Future。

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

实现§

源代码 §

impl Notified<'_>

源代码

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());
        }
    }
}