Custom future not completing

Trying to implement a custom future. Could someone please tell me if anything is obviously wrong? I see the poll() called twice, once when the .await is called and I see the poll() being called after the wake(). But poll() never seems to return Ready.

use std::{    
    future::Future, 
    pin::Pin, 
    task::{Context, Poll, Waker}
};
use std::sync::RwLock;
use crate::io::Buffer;

struct InnerRpcFuture
{
    buffer_ : Option<Buffer>, 
    waker_ : Option<Waker>   
}

impl InnerRpcFuture
{       
    fn wake_me(&self) -> ()
    {
        self.waker_.as_ref().unwrap().wake_by_ref();
    }

    fn is_ready(&self) -> bool
    {
        return self.buffer_.is_some();
    }
}

pub struct RpcFuture
{
    id_ : String,
    inner_ : RwLock<InnerRpcFuture>
}   

impl RpcFuture
{
    pub fn new() -> Self
    {
        return RpcFuture {
            id_ : uuid::Uuid::new_v4().to_string(),
            inner_ : RwLock::new(InnerRpcFuture{
                buffer_ : None, 
                waker_ : None
            })       
        }
    }   

    fn is_ready(&self) -> bool
    {  
        let inner = self.inner_.read().unwrap();
        return inner.is_ready();        
    }

    fn get_state(&self) -> Buffer 
    {                
        let mut inner = self.inner_.write().unwrap();
        return inner.buffer_.take().unwrap();
    }
    
    fn set_waker(&self, waker : Waker) -> ()
    {    
        let mut inner = self.inner_.write().unwrap();
        inner.waker_ = Some(waker);        
    }

    pub fn set(&self, buffer : Buffer) -> ()
    {            
        let mut inner = self.inner_.write().unwrap();
        inner.buffer_ = Some(buffer);         
        println!("waking now {} {}", inner.is_ready(), self.id_);     
        inner.wake_me();
    }
}

impl Future for &RpcFuture
{
    type Output = Buffer;

    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output>
    {
        println!("poll called {} {}", self.is_ready(), self.id_);
        let fut = Pin::into_inner(self);
        if fut.is_ready()
        {            
            return Poll::Ready(fut.get_state());
        }
        else 
        {            
            let waker = cx.waker().clone();
            fut.set_waker(waker);               
            return Poll::Pending;   
        }        
    }
}

How do you spawn the Future and how do you wake it?
By the looks of it set never gets called as your will be waiting for poll to complete.

How does your RPC system work? If it relies on network calls, then it should not be a leaf future; it should delegate calls to other futures that read/write to the network.

In order for "leaf futures" to work, something somewhere has to be running an event loop or getting system callbacks and then (perhaps indirectly) calling the waker. That doesn't just happen on its own. Tokio comes with an event loop and a set of Tokio-specific leaf futures that know how to hook into it. Async-std leaf futures spin up a worker thread with an event loop that triggers the futures to call the waker for whatever executor they're running under. You would need something along those lines to get a leaf future of your own to work.

This topic was automatically closed 90 days after the last reply. We invite you to open a new topic if you have further questions or comments.