(14.2.0)Future only pulled twice.

[unpluggedcoder]

Hi, I know the version 15.0 is in the progress, but I still work on 14.2.0 which depends on futures 0.1.29.

I wrote a AsyncResponse which implement future::Future, and it wait response from tokio::sync::mpsc::channel.
The problem is that it only polled twice when response is not ready.

#[derive(Debug)]
struct AsyncResponse {
    recv: Receiver<usize>,
}

impl RpcFuture for AsyncResponse {
    type Item = Result<String>;
    type Error = types::Error;

    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
        match self.recv.try_recv() {
            Err(TryRecvError::Empty) => {
                info!("AsyncResponse not ready.");
                Ok(Async::NotReady)
            }
            Err(TryRecvError::Closed) => {
                info!("AsyncResponse closed.");
                Err(Error::new(ErrorCode::InternalError).into())
            }
            Ok(value) => {
                info!("AsyncResponse ready, response value: {}", value);
                Ok(Async::Ready(Ok(value.to_string())))
            }
        }
    }
}

#[derive(Default, Clone)]
pub struct Meta(pub Option<UnboundedSender<(usize, Sender<usize>)>>);
impl Metadata for Meta {}

#[rpc(server)]
pub trait Rpc {
    type Metadata;
    /// Performs asynchronous operation
    #[rpc(meta, name = "beFancy")]
    fn call(&self, meta: Self::Metadata) -> BoxFuture<Result<String>>;
}

#[derive(Default, Clone)]
struct RpcImpl;

impl Rpc for RpcImpl {
    type Metadata = Meta;

    fn call(&self, meta: Self::Metadata) -> BoxFuture<Result<String>> {
        let id = REQ_ID.fetch_add(1, Ordering::SeqCst);
        let (tx, rx) = mpsc::channel::<usize>(1);
        if let Some(sender) = meta.0 {
            sender.send((id, tx)).unwrap();
        }
        Box::new(AsyncResponse { recv: rx })
    }
}

fn main() {
    env::set_var("RUST_LOG", "info");
    pretty_env_logger::init();

    let mut rt = runtime::Builder::new()
        .threaded_scheduler()
        .enable_all()
        .build()
        .expect("Runtime build failed.");

    let (broker_sender, mut broker_receiver) = mpsc::unbounded_channel();

    {
        let sender = broker_sender.clone();
        rt.spawn(async {
            let mut io = MetaIoHandler::default();
            let rpc = RpcImpl;

            io.extend_with(rpc.to_delegate());

            let _server = ServerBuilder::new(io)
                .meta_extractor(move |_: &hyper::Request<hyper::Body>| {
                    info!("Meta extractor called.");
                    Meta(Some(sender.clone()))
                })
                .start_http(&"127.0.0.1:9527".parse().unwrap())
                .expect("Unable to start RPC server");

            _server.wait();
        });
    }

    rt.block_on(async move {
        let mut rpc_resps: HashMap<usize, Sender<usize>> = HashMap::new();
        info!("Borker loop start...");
        loop {
            if let Some((id, mut sender)) = broker_receiver.recv().await {
                info!("Broker received: id({}).", id);

                // Sleep for awhile
                thread::sleep(Duration::from_secs(2));

                sender.send(id * id).await.unwrap();
                info!("Broker sent: id({})", id);
                rpc_resps.insert(id, sender);
            } else {
                info!("Broker channel broken.");
                break;
            }
        }
        info!("Broker loop finished.");
    });
}

The broker loop should send reply from other broker channel events, but for the simplicity, it just sleep for awhile.
Output shows that AsyncResponse can receive succeed ONLY if broker send immediately(without sleep).

No sleep:

 INFO  jrpc > Borker loop start...
 INFO  jrpc > Meta extractor called.
 INFO  jrpc > AsyncResponse not ready.
 INFO  jrpc > Broker received: id(0).
 INFO  jrpc > Broker sent: id(0)
 INFO  jrpc > AsyncResponse ready, response value: 0

Wait for awhile:

 INFO  jrpc > Borker loop start...
 INFO  jrpc > Meta extractor called.
 INFO  jrpc > AsyncResponse not ready.
 INFO  jrpc > Broker received: id(0).
 INFO  jrpc > AsyncResponse not ready.
 INFO  jrpc > Broker sent: id(0)
^C

No more poll function called, the future just hang in there.

Full code can be found here.

Am I doing anything wrong?

Second question is that the example codes show Rpc trait method should return Box::pin for FutureResult, but it get compile error, say it should pin in the heap:

70 |     fn call(&self, meta: Self::Metadata) -> BoxFuture<Result<String>> {
   |                                             ------------------------- expected `std::boxed::Box<(dyn futures::future::Future<Error = jsonrpc_core::types::error::Error, Item = std::result::Result<std::string::String, jsonrpc_core::types::error::Error>> + std::marker::Send + 'static)>` because of return type
...
76 |         Box::pin(AsyncResponse { recv: rx })
   |         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
   |         |
   |         expected struct `std::boxed::Box`, found struct `std::pin::Pin`
   |         help: store this in the heap by calling `Box::new`: `Box::new(Box::pin(AsyncResponse { recv: rx }))`
   = note: expected struct `std::boxed::Box<(dyn futures::future::Future<Error = jsonrpc_core::types::error::Error, Item = std::result::Result<std::string::String, jsonrpc_core::types::error::Error>> + std::marker::Send + 'static)>`
              found struct `std::pin::Pin<std::boxed::Box<AsyncResponse>>`
   = note: for more on the distinction between the stack and the heap, read https://doc.rust-lang.org/book/ch15-01-box.html, https://doc.rust-lang.org/rust-by-example/std/box.html, and https://doc.rust-lang.org/std/boxed/index.html

Any help will be appreciate!

[tomusdrw]

Every time you return NotReady, the future should register to a waker, that will be triggered at some point to indicate that the task should be woken up (and polled). My guess is that try_recv does not register a waker, have you tried recv() or simply Stream::poll()?

Second question is that the example codes show Rpc trait method should return Box::pin for FutureResult, but it get compile error, say it should pin in the heap:

The example is for 15.0 most likely, in 14.x you just need to Box::new not Box::pin.

[unpluggedcoder]

The recv function is an async function. I can't await it in the poll method.

According to the docs, I thought it will register automatically when NotReady returned.

Second question. Thanks!

[tomusdrw]

According to the docs, I thought it will register automatically when NotReady returned.

This is trait docs, so it should be treated as implementation suggestions - you implement poll function manually.

The recv function is an async function. I can't await it in the poll method.

Can you point me to the exact function your code is using? I can see try_recv there and then you match on a Result, so it seems like it's a function that is callable outside of task context, like this one:
https://docs.rs/futures/0.1.29/futures/sync/oneshot/struct.Receiver.html#method.try_recv
which means it won't register a waker.

[unpluggedcoder]

Full code can be found here, and it's runnable.

I use tokio::sync::mpsc::channel.

you implement poll function manually.

You mean I should register a waker manually?
I didn't know that before, I'll do some search and learn. Thanks.

[tomusdrw]

In the example you are using tokio=0.2 which isn't compatible with futures=0.1. The try_recv method is simply synchronous, it is meant to be run outside of async context, so it doesn't schedule waker.

You should rather be using something like Stream::poll_next from the Receiver: https://docs.rs/tokio/0.2.21/tokio/sync/mpsc/struct.Receiver.html#trait-implementations
bu this requires futures=0.3 Context object, so instead you should convert the Receiver into Compat (so that it works with 0.1) and then use futures01::Stream::poll method to get the next element.

Closing this issues, since it's not related to the library at all.

[unpluggedcoder]

Thanks for your help! Very appreciate!

[unpluggedcoder]

Here is the compatibility adaptor for using tokio=0.2 with future=0.1.

Add futures=0.3 with compat feature in the Cargo.toml dependencies.

[dependencies]
tokio = { version = "0.2", features = ["full"] }  # tokio 0.2 not compatibile with futures 0.1
futures = { version = "0.3", features = ["compat"] }  # We need "compat" feature.

And turn recv async function into TryFuture with codes below:

use futures::future::{FutureExt, TryFutureExt};

#[derive(Debug)]
struct AsyncResponse<T> {
    recv: Receiver<T>,
}

impl<T: Send + 'static> AsyncResponse<T> {
    fn compat(mut self) -> BoxFuture<Result<T>> {
        let fut = async move {
            match self.recv.recv().await {
                Some(id) => Ok(id),
                None => Err(types::Error::new(types::ErrorCode::InternalError)),
            }
        };
        let compat_fut = fut
            .unit_error()
            .map_err(|_| types::Error::new(types::ErrorCode::InternalError))
            .boxed()
            .compat();
        Box::new(compat_fut)
    }
}

Then use it in the Rpc method:

impl Rpc for RpcImpl {
    type Metadata = Meta;

    fn call(&self, meta: Self::Metadata) -> BoxFuture<Result<usize>> {
        let id = REQ_ID.fetch_add(1, Ordering::SeqCst);
        let (tx, rx) = mpsc::channel::<usize>(1);
        if let Some(sender) = meta.0 {
            sender.send((id, tx)).unwrap();
        }
        let async_resp = AsyncResponse { recv: rx };
        async_resp.compat()
    }
}

Full code can be found HERE.

[unpluggedcoder]

The code above will turn Err(e)(in the async) into "result": {"Err": ...} RPC's response.

I change it to return BoxFuture<T> directly, and code could be more simple.

use jsonrpc_core::{self, types, BoxFuture, MetaIoHandler, Metadata, Result};
use futures::future::{FutureExt, TryFutureExt};

fn async_wait<T: Send + 'static>(
    fut: impl Future<Output = Result<T, types::Error>> + Send + 'static,
) -> BoxFuture<T> {
    let compat_fut = fut.boxed().compat();
    Box::new(compat_fut)
}


impl Rpc for RpcImpl {
    type Metadata = Meta;

    fn call(&self, meta: Self::Metadata) -> BoxFuture<usize> {
        let id = REQ_ID.fetch_add(1, Ordering::SeqCst);
        let (tx, mut rx) = mpsc::channel::<usize>(1);
        if let Some(sender) = meta.0 {
            sender.send((id, tx)).unwrap();
        }
      
        let resp_fut = async move {
            match rx.recv().await {
                Some(id) => Ok(id),
                None => Err(types::Error::new(types::ErrorCode::InternalError).into()),  // Error will be handled right.
            }
        };

        async_wait(resp_fut)    // Stuck here.
    }
}

But I found that both codes will not turn into real async request. The server will stuck on the async_wait, subsequent requests will not be processed until async finished.

[tomusdrw]

@unpluggedcoder which server are you using? Could you give a runnable code sample? Note that the async doesn't mean you can process infinite number of things concurrently, depending on the set up there is a fixed number of threads so it's fine to spawn infinite number of tasks, but only a fixed number of them will progress in parallel.

[unpluggedcoder]

Thanks for the reply!
I put the runnable code in my repo:

https://github.com/unpluggedcoder/rust-jsonrpc-example

[tomusdrw]

@unpluggedcoder sorry to get back so late, I've finally found some time to take a look at your example. The error seems to be in the TCP server implementation, where each request coming from a single session is processed sequentially. Started a separate issue for this: #577
Thanks again for your detailed example!