Questions on futures/async/await roadmap

[phlip9]

Hey guys! I'm looking to use rust-libp2p in a project and have some observations + questions.

First, some initial observations after playing around with it for a few weeks:

1. I'm super excited by rust-libp2p and see its potential, particularly if we can improve the same underlying framework and then easily share protocols built using the framework.
2. After implementing some basic protocols with rust-libp2p, I find the "low-level" components, specifically Transport, Upgrade, and StreamMuxer, most appealing. The composability, "swap-ability", etc is nice.
3. I like the idea of Swarm, NetworkBehaviour, and ProtocolsHandler though the actual API is a bit clunky and could probably use some iteration. Manually implementing state machines in NetworkBehaviour or ProtocolsHandler using the poll API seems overly mechanical and error-prone.

Next, some questions about direction:

1. futures v0.3.0: Is there a plan to begin migration to the new futures API? For the cost of some Box indirection and compat() wrapping, we could probably begin an incremental, piece-wise migration sooner rather than later.
2. async/await: Similarly, is there a plan to adopt async/await? Are we waiting for stabilization? I think there is great potential to simplifiy much of the manual state machine logic in e.g. ping/protocol.rs:92 or kad/kandler.rs:570. For my own understanding, I tried quickly sketching out Ping's OutboundUpgrade using async/await:

impl<TSocket> OutboundUpgrade<TSocket> for Ping
where
    TSocket: AsyncRead + AsyncWrite,
{
    async fn upgrade_outbound(self, socket: TSocket, _: Self::Info) -> Result<Duration, io::Error> {
        use tokio_io::io::{flush, read_exact, shutdown, write_all};

        let payload = EntropyRng::default().sample(Standard);

        // send and flush nonce
        let (socket, send_payload) = await!(write_all(socket, payload))?;
        let socket = await!(flush(socket))?;
        let started = Instant::now();

        // wait for response, check nonces match, and record latency.
        let (socket, recv_payload) = await!(read_exact(socket, [0; 32]))?;
        let ping_time = started.elapsed();
        if recv_payload != send_payload {
            return Err(io::Error::new(/* .. */));
        }

        let _ = await!(shutdown(socket))?;

        Ok(ping_time)
    }
}

3. API: Would it make sense to separate modules in the NetworkBehaviour/ProtocolsHandler hierarchy by "hard" channel boundaries, like even closer to an actor-ish model? Instead of owning your sub-behaviour directly, you hold an in-event sender to push events into it and an out-event receiver to pull events it generates. Likewise for your parent NetworkBehaviour or Swarm. I think (?) this might make the async/await transition easier? Your async poll equivalent would select! across the receivers in a loop and handle events that way. The tricky part would be figuring out how to ergonomically send events on the senders if they are bounded queues. Each behaviour would get spawned as its own task on the executor. With the current approach, my uninformed and not-data-driven worry is that having one giant Task encapsulate the whole NetworkBehaviour hierarchy might be too coarse-grained.
4. Communication: Finally, what is the best communication channel? Github Issues? Discord? IRC?

Some of these points could probably live in their own issue. Look forward to hearing your thoughts.

Thanks!
-Philip

[tomaka]

Hey 👋

I like the idea of Swarm, NetworkBehaviour, and ProtocolsHandler though the actual API is a bit clunky and could probably use some iteration. Manually implementing state machines in NetworkBehaviour or ProtocolsHandler using the poll API seems overly mechanical and error-prone.

I agree with that feeling. However "could probably use some iteration" is easier said than done 😅

futures v0.3.0: Is there a plan to begin migration to the new futures API? For the cost of some Box indirection and compat() wrapping, we could probably begin an incremental, piece-wise migration sooner rather than later.

Not any time soon. We've put a lot of efforts to not use Boxes unless necessary, in order to avoid having to decide whether or not to require the Send/Sync bounds.
@gnunicorn has been playing a bit with the new futures and the backwards compatibility layer so maybe he has something to say.

async/await: Similarly, is there a plan to adopt async/await? Are we waiting for stabilization?

It is certain that async/await will greatly improve things, and probably divide by 2 the number of lines of code of this library (and same for impl Trait).
However the pessimistic side of me thinks that this feature is not going to be stable any time soon (maybe not this year), and we are pinned to the stable Rust.

Instead of owning your sub-behaviour directly, you hold an in-event sender to push events into it and an out-event receiver to pull events it generates.
Each behaviour would get spawned as its own task on the executor.

That is in fact already the case. The node handlers and protocols handlers run on separate tasks. When they emit events, they are propagated to the "main" task that only runs the network behaviour.

Communication: Finally, what is the best communication channel? Github Issues? Discord? IRC?

cc #746
There's no agreed upon channel yet.

[phlip9]

That is in fact already the case. The node handlers and protocols handlers run on separate tasks. When they emit events, they are propagated to the "main" task that only runs the network behaviour.

Ahh I seem to have misunderstood. Thanks for pointing that out!

I also have some additional follow-up questions:

Some of us are (understandably) worried about pulling in a core dependency without any of us being core contributors. If we were to pull in rust-libp2p, we would likely invest heavily in it, but some of us are concerned about commit rights and the pull request review process. I'm curious what your thoughts are on commit rights etc?

[tomaka]

Some of us are (understandably) worried about pulling in a core dependency without any of us being core contributors. If we were to pull in rust-libp2p, we would likely invest heavily in it, but some of us are concerned about commit rights and the pull request review process. I'm curious what your thoughts are on commit rights etc?

That's quite a broad question.

EDIT: I originally put my opinion here, but I guess it doesn't really matter.

Right now we're merging changes quite quickly in this repo because all the main contributors are within Parity and we can discuss changes in real life in our office. But if more people start opening pull requests, we'll probably be more cautious when it comes to the merging speed.

I don't know what the policy of Protocol Labs (the owners of the libp2p organization) when it comes to giving merge rights is. cc @mgoelzer @raulk

[ghost]

Some of us are (understandably) worried about pulling in a core dependency without any of us being core contributors. If we were to pull in rust-libp2p, we would likely invest heavily in it, but some of us are concerned about commit rights and the pull request review process. I'm curious what your thoughts are on commit rights etc?

Building on what @tomaka is saying:

The current governance model of rust-libp2p is basically that @tomaka and @twittner are the maintainers. This reflects how the project was developed historically. If you want to invest heavily in rust-libp2p in the sense of actually contributing code back into the project (add features, etc), then, yeah, I absolutely think you and your team could become core contributors. I'm assuming that is the sense you meant it because you're talking about pull request review and such.

Maybe you and your group, @tomaka, myself and @raulk should talk more about where you're going and what you want to contribute? You can email me at mike@libp2p.io, or ping @mgoelzer in https://riot.im/app/#/room/#libp2p:matrix.org. Email might be easier to do a long discussion over time.

Excited that you're interested and hope to chat further @phlip9.

[tomaka]

The issue went a bit off-topic, as it was originally about async/await. Let's recenter.

Since futures will likely get stabilized in Rust 1.35, we should consider transitioning to standard-library-futures (also known as futures 0.3 and futures-preview) as a first step.

The blocker on this is that we either need tokio to transition to new futures, or a stable way to turn old futures into new futures.

Considering that the standard-library doesn't have an equivalent to Stream, another step would also be to replace the Stream implementations with something else. For example Swarm could instead have a method next_event(&mut self) -> impl Future<Output = ...>.

The various inherent poll() methods scattered around the library can also be replaced with a next_event() method, in order to become usable with await!.
Theoretically we could already start working on that even before futures get stabilized; unfortunately futures 0.1 must always return a Result, and it's not great to turn methods that return T into methods that return Result<T, Void>.

[boomshroom]

Futures 0.3 are stabilized in the current nightly and will be in stable 1.35. async/await will be coming later and while they would make things much simpler, they aren't strictly required to start transitioning.

Something to note from the conversation, boxing futures 0.3 is only necessary if they require pinning, and that only occurs due to holding references across yield/await points. Since futures won't initially have async/await, boxing is not necessary except to get a trait object, which would be the case with 0.1 anyways.

[boomshroom]

The blocker on this is that we either need tokio to transition to new futures, or a stable way to turn old futures into new futures.

The compat feature on futures-preview is exactly this, though it may not be as stable as we'd like, mostly because of deprecation of futures 0.1 rather than actual API changes.

Considering that the standard-library doesn't have an equivalent to Stream, another step would also be to replace the Stream implementations with something else. For example Swarm could instead have a method next_event(&mut self) -> impl Future<Output = ...>.

Again, futures-preview does provide a Stream trait and combinators. The reason it isn't in the standard library is just because it isn't strictly necessary to compile async/await.

The various inherent poll() methods scattered around the library can also be replaced with a next_event() method, in order to become usable with await!.

I won't argue this as I agree that returning futures is probably better than providing explicit poll functions, but this is why futures-preview has poll_fn.

Theoretically we could already start working on that even before futures get stabilized; unfortunately futures 0.1 must always return a Result, and it's not great to turn methods that return T into methods that return Result<T, Void>.

Are you expecting me to give something from futures-preview that handles this? Good you're catching on. unit_error() It would be nice for the error type to be Infallible or !, but it's to accommodate existing tokio infrastructure.

[EDIT] 5 hours before I posted this, futures-preview added never_error as I mentioned wanting.

[dignifiedquire]

the new futures just landed on stable rust: https://blog.rust-lang.org/2019/07/04/Rust-1.36.0.html

[tomaka]

Some random thoughts:

Pass the executor by parameter

The executor is no longer a thread-local thingy and should instead be passed explicitly.
I think we should require the executor to be passed explicitly in the low-level code, but default to a ThreadsPool (with the option to customize) in the higher-level code.

Protocol upgrades

It should be possible to write something like:

let upgrade = upgrade::from_fn("/foo/1.0.0", async |stream| {
    let handshake_msg = read_one(&stream).await?;
    check(&handshake_msg)?;
    Ok(stream)
});

And the returned object could implement OutboundUpgrade and InboundUpgrade.

Node handler

The NodeHandler trait can either generate events or open substreams. It then receives an event when a substream is open, or an event received from the outside.

We could similarly do something like that:

let handler = handler::from_fn(async |context| {
    let substream = context.open_substream().await;

    loop {
        let in_event = context.next_event().await;
        context.generate_event(..).await;
    }
});

And same for the ProtocolHandler and NetworkBehaviour.

[tomaka]

Even though I opened #1196, we might not switch to new futures before async/await just yet?

The parts that use tokio-codec, copying between read and write, and sink need to be rewritten.
However since the new ecosystem of futures is built around non-'static futures (which work great only if you have async/await), it would quite a large effort that would be avoided by just waiting for async/await to be stable.

EDIT: of course if someone else wants to do it, you're welcome! But it's a lot of efforts and many parts of it might be deleted next month when async/await are stable.

[tomaka]

I tried playing a bit with async/await, and we can probably reduce by two thirds some elements of the code of rust-libp2p. Here's an example diff:

diff --git a/core/src/upgrade/apply.rs b/core/src/upgrade/apply.rs
index d9a3e7a1..787debdf 100644
--- a/core/src/upgrade/apply.rs
+++ b/core/src/upgrade/apply.rs
@@ -41,16 +41,15 @@ where
 }
 
 /// Tries to perform an upgrade on an inbound connection or substream.
-pub fn apply_inbound<C, U>(conn: C, up: U) -> InboundUpgradeApply<C, U>
+pub async fn apply_inbound<C, U>(conn: C, up: U) -> Result<U::Output, UpgradeError<U::Error>>
 where
     C: AsyncRead + AsyncWrite,
     U: InboundUpgrade<C>,
 {
     let iter = UpgradeInfoIterWrap(up);
-    let future = multistream_select::listener_select_proto(conn, iter);
-    InboundUpgradeApply {
-        inner: InboundUpgradeApplyState::Init { future }
-    }
+    let future = multistream_select::listener_select_proto(conn, iter).compat();
+    let (info, connection, upgrade) = future.await?;
+    Ok(upgrade.0.upgrade_inbound(connection, info.0).compat().await.map_err(UpgradeError::Apply)?)
 }
 
 /// Tries to perform an upgrade on an outbound connection or substream.
@@ -66,75 +65,6 @@ where
     }
 }
 
-/// Future returned by `apply_inbound`. Drives the upgrade process.
-pub struct InboundUpgradeApply<C, U>
-where
-    C: AsyncRead + AsyncWrite,
-    U: InboundUpgrade<C>
-{
-    inner: InboundUpgradeApplyState<C, U>
-}
-
-enum InboundUpgradeApplyState<C, U>
-where
-    C: AsyncRead + AsyncWrite,
-    U: InboundUpgrade<C>
-{
-    Init {
-        future: ListenerSelectFuture<C, UpgradeInfoIterWrap<U>, NameWrap<U::Info>>,
-    },
-    Upgrade {
-        future: U::Future
-    },
-    Undefined
-}
-
-impl<C, U> Future for InboundUpgradeApply<C, U>
-where
-    C: AsyncRead + AsyncWrite,
-    U: InboundUpgrade<C>,
-{
-    type Item = U::Output;
-    type Error = UpgradeError<U::Error>;
-
-    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
-        loop {
-            match mem::replace(&mut self.inner, InboundUpgradeApplyState::Undefined) {
-                InboundUpgradeApplyState::Init { mut future } => {
-                    let (info, connection, upgrade) = match future.poll()? {
-                        Async::Ready(x) => x,
-                        Async::NotReady => {
-                            self.inner = InboundUpgradeApplyState::Init { future };
-                            return Ok(Async::NotReady)
-                        }
-                    };
-                    self.inner = InboundUpgradeApplyState::Upgrade {
-                        future: upgrade.0.upgrade_inbound(connection, info.0)
-                    };
-                }
-                InboundUpgradeApplyState::Upgrade { mut future } => {
-                    match future.poll() {
-                        Ok(Async::NotReady) => {
-                            self.inner = InboundUpgradeApplyState::Upgrade { future };
-                            return Ok(Async::NotReady)
-                        }
-                        Ok(Async::Ready(x)) => {
-                            debug!("Successfully applied negotiated protocol");
-                            return Ok(Async::Ready(x))
-                        }
-                        Err(e) => {
-                            debug!("Failed to apply negotiated protocol");
-                            return Err(UpgradeError::Apply(e))
-                        }
-                    }
-                }
-                InboundUpgradeApplyState::Undefined =>
-                    panic!("InboundUpgradeApplyState::poll called after completion")
-            }
-        }
-    }
-}
-
 /// Future returned by `apply_outbound`. Drives the upgrade process.
 pub struct OutboundUpgradeApply<C, U>
 where

[tomaka]

Some updating, for people who may be following this issue: we now have a branch named stable-futures containing the work-in-progress for updating this library to stable futures.

[tomaka]

#1328 has been merged. Let's close this.