Polling a future from other futures

I will start this question with a bit of context: I am trying to find a clean solution to the implementation of an object that represents some sort of remote device which I can interact with using its associated async functions.

For example:

let remote_device = RemoteDevice::new((127,0,0,1), 1234);
let id = remote_device.id().await;

Where id is an async function associated with the RemoteDevice that will send a request over the network and will complete with either an error or the identifier of the remote device, that is:

impl RemoteDevice {
    pub async fn id(&self) -> Result<String, Error> {...}
    // other functions

I would like to design this object so that multiple requests can be made simulatenously. To do this, I add an mpsc channel to the object that takes a request and an optional oneshot channel in case a response is required as a tuple.

I then add an addition field demultiplexer to hold a future that will multiplex requests on to the underlying transport and demultiplex responses, sending them back to the async methods. In this case, the definition of RemoteDevice may look something like:

struct RemoteDevice {
   requests: mpsc::UnboundedSender<(Request, Option<oneshot::Sender<Response>>)>,
   demultiplexer: Box<dyn Future<Output = ()>>,
   // other fields

With this in place, I can write an async function for my remote device as follows:

pub async fn id(&self) -> Result<Response, RecvError> {
   let (response_tx, response_rx) = oneshot::channel();
   self.requests.send((Request::GetId, Some(response_tx)));

The problem with this solution so far is that awaiting response_rx will not poll demultiplexer. Furthermore, I can't poll demultiplexer from id (e.g., via join or select) without having id take a mutable reference to demultiplexer, which I would like to avoid this since being able to call different methods on this object concurrently is part of the design criteria. The semantics that I would like to have are that whenever something is awaiting one of the futures generated from the async functions (e.g. the id function), demultiplexer is also polled (since otherwise the other futures will never complete).

Preferably I would like to work at the async/await level and not drop down to manual implementations of the Future trait. I feel like there must be a standard pattern/solution for this problem, however, I have no idea what I would search for. Any suggestions?

Just to confirm my understanding, demultiplexer will receive messages sent on requests, handling them internally? If so, I recommend spawning it instead, going for an actor.

1 Like

Yes, demultiplexer also owns the receiver end of the requests mpsc channel and sends the received requests over the network.

Spawning it is definitely a solution, the only thing that makes this not so nice is that, as far as I am aware, I then have to tie my code to a particular executor, right?

Or I guess if I really didn't want to do this, I could also write an async function called run which needs to be passed to an executor for the other async functions to work -- which I am also not the biggest fan of since it makes the code less contained...

I am just looking at the page that you wrote about actors now. This is exactly what I am looking for, I'll read it thoroughly soon.

One technique that libraries sometimes use is to return the future to the user and ask them to spawn it. This lets you be independent of the executor in use.

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.