Hi,

Context for my question: I'm writing a Future that will receive a completion notification on another thread (due to OS API constraints). My plan is to pass the Waker to that thread and wake() from there. I'm wondering if this is correct usage of the Waker.

Generally, I understand that an executor is allowed to poll() a pending Future at any time, even if the future didn't signal a Waker. What I'm missing is what must a pending Future do if it receives a different Waker in the poll().

I assume the Future should ideally replace the previous Waker with the current Waker, and use the latest Waker for notification, but is it required?
In other words, do older Wakers remain operational during the entire life of the future?

For example, is this dummy code correct or can it get stuck because the executor may not poll the future due to using an "expired" Waker?

#[derive(Default)]
struct SleepyFuture {
    thread: Option<thread::JoinHandle<()>>,
}

impl Future for SleepyFuture {
    type Output = ();

    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
        let thread = self.thread.get_or_insert_with(|| {
            // Capture only the first ever waker.
            let waker = cx.waker().clone();
            thread::spawn(move || {
                waker.wake_by_ref();
                thread::sleep(Duration::from_secs(1));
                waker.wake_by_ref();
                thread::sleep(Duration::from_secs(1));
                waker.wake_by_ref();
                thread::sleep(Duration::from_secs(1));
                waker.wake_by_ref();
            })
        });

        // Here we ignore the current Waker passed to poll(), relying on the
        // thread to call the original waker.
        match thread.is_finished() {
            false => Poll::Pending,
            true => Poll::Ready(()),
        }
    }
}

(This code doesn't get stuck in Tokio, but I'm concerned it's due to current implementation details.)

3 posts - 2 participants

Read full topic