Should async-byte-channel work with a Tokio LocalSet?

[JonasVautherin]

I can successfully use async-byte-channel with futures::executor::LocalPool, like in the tests, but somehow it blocks when I use them with Tokio and a LocalSet.

I wanted to try with a multi-threaded Tokio executor, but RpcSystem is not Send, so apparently it's not possible.

Should I be able to use async-byte-channel with a Tokio LocalSet, or did I miss something that makes it non-cooperative and therefore failing on a single thread executor?

[JonasVautherin]

Note that I tried using Tokio's DuplexStream instead, but it blocks just the same 😕.

[dwrensha]

I expect it to work with any executor. Can you post your code?

[JonasVautherin]

EDIT: I was missing a line, see the next answer 🤦‍♂️.

Sure. I have a simple test interface that implements echo, and a test for it.

use capnp::Error;
use capnp::capability::Promise;
use crate::test_capnp::test_interface;
use capnp_rpc::pry;

pub struct TestImpl;

impl test_interface::Server for TestImpl {
    fn echo(
        &mut self,
        params: test_interface::EchoParams,
        mut results: test_interface::EchoResults,
    ) -> Promise<(), Error> {
        let input = pry!(pry!(params.get()).get_input()).to_string().unwrap();
        results.get().set_output(input);

        Promise::ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use capnp_rpc::{RpcSystem, twoparty, rpc_twoparty_capnp};
    use crate::test::TestImpl;
    use crate::test_capnp;
    use tokio::io::duplex;
    use tokio::runtime::UnhandledPanic;
    use tokio::time::{sleep, Duration};
    use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};

    #[tokio::test]
    async fn it_echoes_successfully() {
        let mut local_task_set = tokio::task::LocalSet::new();
        local_task_set.unhandled_panic(UnhandledPanic::ShutdownRuntime);

        let (client_writer, server_reader) = async_byte_channel::channel();
        let (server_writer, client_reader) = async_byte_channel::channel();

        // Create server_rpc_system
        let server_network = Box::new(twoparty::VatNetwork::new(
            server_reader,
            server_writer,
            rpc_twoparty_capnp::Side::Server,
            Default::default(),
        ));
        let test_impl: test_capnp::test_interface::Client = capnp_rpc::new_client(TestImpl {});
        let server_rpc_system = RpcSystem::new(server_network, Some(test_impl.client));
 
        // Create client_rpc_system
        let client_network = Box::new(twoparty::VatNetwork::new(
            client_reader,
            client_writer,
            rpc_twoparty_capnp::Side::Client,
            Default::default(),
        ));
        let mut client_rpc_system = RpcSystem::new(client_network, None);
   
        // Bootstrap client
        let client: test_capnp::test_interface::Client = client_rpc_system.bootstrap(rpc_twoparty_capnp::Side::Server);

        // Server
        local_task_set.spawn_local(async {
            tokio::task::spawn_local(server_rpc_system);
        });

        // Timeout task
        local_task_set.spawn_local(async {
            sleep(Duration::from_millis(3000)).await;
            panic!("Test timed out!!!");
        });

        // Run the client task
        local_task_set.run_until(async {
            let mut echo_request = client.echo_request();
            echo_request.get().set_input("Hello, World");
            let response = echo_request.send().promise.await.unwrap();
            let output = response.get().unwrap().get_output().unwrap().to_string().unwrap();

            assert_eq!("Hello, World", output);
        }).await;
    }
}

I would expect the test to pass, but instead it times out with:

---- test::tests::it_echoes_successfully stdout ----
thread 'test::tests::it_echoes_successfully' panicked at src/test.rs:70:13:
Test timed out!!!
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace
thread 'test::tests::it_echoes_successfully' panicked at /home/jonas/.cargo/registry/src/index.crates.io-6f17d22bba15001f/tokio-1.37.0/src/task/local.rs:665:13:
a spawned task panicked and the LocalSet is configured to shutdown on unhandled panic

As mentioned above, instead of the async-byte-channel I tried with a DuplexStream, by replacing the client_reader, client_writer, server_reader, server_writer definitions above with:

        // Prepare DuplexStream and convert it for VatNetwork
        let (client_tokio, server_tokio) = duplex(4096);
        let (client_reader_tokio, client_writer_tokio) = tokio::io::split(client_tokio);
        let client_reader = TokioAsyncReadCompatExt::compat(client_reader_tokio);
        let client_writer = TokioAsyncWriteCompatExt::compat_write(client_writer_tokio);
        let (server_reader_tokio, server_writer_tokio) = tokio::io::split(server_tokio);
        let server_reader = TokioAsyncReadCompatExt::compat(server_reader_tokio);
        let server_writer = TokioAsyncWriteCompatExt::compat_write(server_writer_tokio);

But it fails the exact same way: the client doesn't go further than the line let response = echo_request.send().promise.await.unwrap();.

[JonasVautherin]

Argh I found it, I was not calling tokio::task::spawn_local(client_rpc_system); 🤦‍♂️. Adding it to the client task above makes it work with both async-byte-channel and DuplexStream.

Just for my undersatnding: it is intuitive to me that I need to call tokio::task::spawn_local(server_rpc_system); (because the server has to be running, obviously), but why is there a need for the same on the client side?

[dwrensha]

1. The inner write queue of an RPC connection (client or server) is driven by a TaskSet that is owned by the RpcSystem:
   

   capnproto-rust/capnp-rpc/src/lib.rs

   Line 200 in 34c7cd6

   handle.add(network.drive_until_shutdown().then(move |r| {

   
   If you don't run the RpcSystem future, then that TaskSet never gets driven. If we committed to only supporting tokio, then we could maybe internally use tokio's spawn functionality here instead, but I want to remain agnostic to async runtime.

2. In Cap'n Proto RPC, there is no strong distinction between the client side and server side. For example, a "client" might make a call that includes so local capabilities, and then the "server" will be able to later make calls on those capabilities.

[JonasVautherin]

Thanks for the insights! 😊