Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

multistream-select: Remove parallel dialing optimization #2934

Merged
merged 8 commits into from
Sep 29, 2022
Merged

multistream-select: Remove parallel dialing optimization #2934

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
d326bee
fix(multistream-select): remove parallel dialing optimization
dignifiedquire Sep 23, 2022
ee19c59
apply CR
dignifiedquire Sep 23, 2022
0becbbf
Update misc/multistream-select/CHANGELOG.md
thomaseizinger Sep 23, 2022
0e6d27c
Update misc/multistream-select/tests/dialer_select.rs
dignifiedquire Sep 23, 2022
8620a40
Update misc/multistream-select/tests/dialer_select.rs
thomaseizinger Sep 23, 2022
855c918
Update misc/multistream-select/CHANGELOG.md
mxinden Sep 27, 2022
c1e3ffb
Merge branch 'master' into fix-ms-remove-parallel-dial
mxinden Sep 27, 2022
99e0ecc
Merge branch 'master' into fix-ms-remove-parallel-dial
mxinden Sep 29, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions misc/multistream-select/CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,3 +1,7 @@
# Unreleased
thomaseizinger marked this conversation as resolved.
Show resolved Hide resolved

- Remove parallel dialing optimization, to avoid requiring the use of the `ls` command.
thomaseizinger marked this conversation as resolved.
Show resolved Hide resolved

# 0.11.0 [2022-01-27]

- Migrate to Rust edition 2021 (see [PR 2339]).
Expand Down
211 changes: 5 additions & 206 deletions misc/multistream-select/src/dialer_select.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@
use crate::protocol::{HeaderLine, Message, MessageIO, Protocol, ProtocolError};
use crate::{Negotiated, NegotiationError, Version};

use futures::{future::Either, prelude::*};
use futures::prelude::*;
use std::{
convert::TryFrom as _,
iter, mem,
Expand All @@ -39,12 +39,6 @@ use std::{
/// returned `Future` resolves with the name of the negotiated protocol and
/// a [`Negotiated`] I/O stream.
///
/// The chosen message flow for protocol negotiation depends on the numbers of
/// supported protocols given. That is, this function delegates to serial or
/// parallel variant based on the number of protocols given. The number of
/// protocols is determined through the `size_hint` of the given iterator and
/// thus an inaccurate size estimate may result in a suboptimal choice.
///
/// Within the scope of this library, a dialer always commits to a specific
/// multistream-select [`Version`], whereas a listener always supports
/// all versions supported by this library. Frictionless multistream-select
Expand All @@ -55,44 +49,13 @@ pub fn dialer_select_proto<R, I>(
protocols: I,
version: Version,
) -> DialerSelectFuture<R, I::IntoIter>
where
R: AsyncRead + AsyncWrite,
I: IntoIterator,
I::Item: AsRef<[u8]>,
{
let iter = protocols.into_iter();
// We choose between the "serial" and "parallel" strategies based on the number of protocols.
if iter.size_hint().1.map(|n| n <= 3).unwrap_or(false) {
Either::Left(dialer_select_proto_serial(inner, iter, version))
} else {
Either::Right(dialer_select_proto_parallel(inner, iter, version))
}
}

/// Future, returned by `dialer_select_proto`, which selects a protocol and dialer
/// either trying protocols in-order, or by requesting all protocols supported
/// by the remote upfront, from which the first protocol found in the dialer's
/// list of protocols is selected.
pub type DialerSelectFuture<R, I> = Either<DialerSelectSeq<R, I>, DialerSelectPar<R, I>>;

/// Returns a `Future` that negotiates a protocol on the given I/O stream.
///
/// Just like [`dialer_select_proto`] but always using an iterative message flow,
/// trying the given list of supported protocols one-by-one.
///
/// This strategy is preferable if the dialer only supports a few protocols.
pub(crate) fn dialer_select_proto_serial<R, I>(
inner: R,
protocols: I,
version: Version,
) -> DialerSelectSeq<R, I::IntoIter>
where
R: AsyncRead + AsyncWrite,
I: IntoIterator,
I::Item: AsRef<[u8]>,
{
let protocols = protocols.into_iter().peekable();
DialerSelectSeq {
DialerSelectFuture {
version,
protocols,
state: SeqState::SendHeader {
thomaseizinger marked this conversation as resolved.
Show resolved Hide resolved
Expand All @@ -101,39 +64,10 @@ where
}
}

/// Returns a `Future` that negotiates a protocol on the given I/O stream.
///
/// Just like [`dialer_select_proto`] but always using a message flow that first
/// requests all supported protocols from the remote, selecting the first
/// protocol from the given list of supported protocols that is supported
/// by the remote.
///
/// This strategy may be beneficial if the dialer supports many protocols
/// and it is unclear whether the remote supports one of the first few.
pub(crate) fn dialer_select_proto_parallel<R, I>(
inner: R,
protocols: I,
version: Version,
) -> DialerSelectPar<R, I::IntoIter>
where
R: AsyncRead + AsyncWrite,
I: IntoIterator,
I::Item: AsRef<[u8]>,
{
let protocols = protocols.into_iter();
DialerSelectPar {
version,
protocols,
state: ParState::SendHeader {
io: MessageIO::new(inner),
},
}
}

/// A `Future` returned by [`dialer_select_proto_serial`] which negotiates
/// A `Future` returned by [`dialer_select_proto`] which negotiates
/// a protocol iteratively by considering one protocol after the other.
#[pin_project::pin_project]
pub struct DialerSelectSeq<R, I: Iterator> {
pub struct DialerSelectFuture<R, I: Iterator> {
// TODO: It would be nice if eventually N = I::Item = Protocol.
protocols: iter::Peekable<I>,
state: SeqState<R, I::Item>,
Expand All @@ -148,7 +82,7 @@ enum SeqState<R, N> {
Done,
}

impl<R, I> Future for DialerSelectSeq<R, I>
impl<R, I> Future for DialerSelectFuture<R, I>
where
// The Unpin bound here is required because we produce a `Negotiated<R>` as the output.
// It also makes the implementation considerably easier to write.
Expand Down Expand Up @@ -267,138 +201,3 @@ where
}
}
}

/// A `Future` returned by [`dialer_select_proto_parallel`] which negotiates
/// a protocol selectively by considering all supported protocols of the remote
/// "in parallel".
#[pin_project::pin_project]
pub struct DialerSelectPar<R, I: Iterator> {
protocols: I,
state: ParState<R, I::Item>,
version: Version,
}

enum ParState<R, N> {
SendHeader { io: MessageIO<R> },
SendProtocolsRequest { io: MessageIO<R> },
Flush { io: MessageIO<R> },
RecvProtocols { io: MessageIO<R> },
SendProtocol { io: MessageIO<R>, protocol: N },
Done,
}

impl<R, I> Future for DialerSelectPar<R, I>
where
// The Unpin bound here is required because we produce a `Negotiated<R>` as the output.
// It also makes the implementation considerably easier to write.
R: AsyncRead + AsyncWrite + Unpin,
I: Iterator,
I::Item: AsRef<[u8]>,
{
type Output = Result<(I::Item, Negotiated<R>), NegotiationError>;

fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
let this = self.project();

loop {
match mem::replace(this.state, ParState::Done) {
ParState::SendHeader { mut io } => {
match Pin::new(&mut io).poll_ready(cx)? {
Poll::Ready(()) => {}
Poll::Pending => {
*this.state = ParState::SendHeader { io };
return Poll::Pending;
}
}

let msg = Message::Header(HeaderLine::from(*this.version));
if let Err(err) = Pin::new(&mut io).start_send(msg) {
return Poll::Ready(Err(From::from(err)));
}

*this.state = ParState::SendProtocolsRequest { io };
}

ParState::SendProtocolsRequest { mut io } => {
match Pin::new(&mut io).poll_ready(cx)? {
Poll::Ready(()) => {}
Poll::Pending => {
*this.state = ParState::SendProtocolsRequest { io };
return Poll::Pending;
}
}

if let Err(err) = Pin::new(&mut io).start_send(Message::ListProtocols) {
return Poll::Ready(Err(From::from(err)));
}

log::debug!("Dialer: Requested supported protocols.");
*this.state = ParState::Flush { io }
}

ParState::Flush { mut io } => match Pin::new(&mut io).poll_flush(cx)? {
Poll::Ready(()) => *this.state = ParState::RecvProtocols { io },
Poll::Pending => {
*this.state = ParState::Flush { io };
return Poll::Pending;
}
},

ParState::RecvProtocols { mut io } => {
let msg = match Pin::new(&mut io).poll_next(cx)? {
Poll::Ready(Some(msg)) => msg,
Poll::Pending => {
*this.state = ParState::RecvProtocols { io };
return Poll::Pending;
}
// Treat EOF error as [`NegotiationError::Failed`], not as
// [`NegotiationError::ProtocolError`], allowing dropping or closing an I/O
// stream as a permissible way to "gracefully" fail a negotiation.
Poll::Ready(None) => return Poll::Ready(Err(NegotiationError::Failed)),
};

match &msg {
Message::Header(h) if h == &HeaderLine::from(*this.version) => {
*this.state = ParState::RecvProtocols { io }
}
Message::Protocols(supported) => {
let protocol = this
.protocols
.by_ref()
.find(|p| supported.iter().any(|s| s.as_ref() == p.as_ref()))
.ok_or(NegotiationError::Failed)?;
log::debug!(
"Dialer: Found supported protocol: {}",
String::from_utf8_lossy(protocol.as_ref())
);
*this.state = ParState::SendProtocol { io, protocol };
}
_ => return Poll::Ready(Err(ProtocolError::InvalidMessage.into())),
}
}

ParState::SendProtocol { mut io, protocol } => {
match Pin::new(&mut io).poll_ready(cx)? {
Poll::Ready(()) => {}
Poll::Pending => {
*this.state = ParState::SendProtocol { io, protocol };
return Poll::Pending;
}
}

let p = Protocol::try_from(protocol.as_ref())?;
if let Err(err) = Pin::new(&mut io).start_send(Message::Protocol(p.clone())) {
return Poll::Ready(Err(From::from(err)));
}

log::debug!("Dialer: Expecting proposed protocol: {}", p);
let io = Negotiated::expecting(io.into_reader(), p, None);

return Poll::Ready(Ok((protocol, io)));
}

ParState::Done => panic!("ParState::poll called after completion"),
}
}
}
}
35 changes: 1 addition & 34 deletions misc/multistream-select/src/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,7 +22,6 @@

#![cfg(test)]
thomaseizinger marked this conversation as resolved.
Show resolved Hide resolved

use crate::dialer_select::{dialer_select_proto_parallel, dialer_select_proto_serial};
use crate::{dialer_select_proto, listener_select_proto};
use crate::{NegotiationError, Version};

Expand Down Expand Up @@ -182,38 +181,6 @@ fn negotiation_failed() {
}
}

#[test]
fn select_proto_parallel() {
async fn run(version: Version) {
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let listener_addr = listener.local_addr().unwrap();

let server = async_std::task::spawn(async move {
let connec = listener.accept().await.unwrap().0;
let protos = vec![b"/proto1", b"/proto2"];
let (proto, io) = listener_select_proto(connec, protos).await.unwrap();
assert_eq!(proto, b"/proto2");
io.complete().await.unwrap();
});

let client = async_std::task::spawn(async move {
let connec = TcpStream::connect(&listener_addr).await.unwrap();
let protos = vec![b"/proto3", b"/proto2"];
let (proto, io) = dialer_select_proto_parallel(connec, protos.into_iter(), version)
.await
.unwrap();
assert_eq!(proto, b"/proto2");
io.complete().await.unwrap();
});

server.await;
client.await;
}

async_std::task::block_on(run(Version::V1));
async_std::task::block_on(run(Version::V1Lazy));
}

#[test]
fn select_proto_serial() {
thomaseizinger marked this conversation as resolved.
Show resolved Hide resolved
async fn run(version: Version) {
Expand All @@ -231,7 +198,7 @@ fn select_proto_serial() {
let client = async_std::task::spawn(async move {
let connec = TcpStream::connect(&listener_addr).await.unwrap();
let protos = vec![b"/proto3", b"/proto2"];
let (proto, io) = dialer_select_proto_serial(connec, protos.into_iter(), version)
let (proto, io) = dialer_select_proto(connec, protos.into_iter(), version)
.await
.unwrap();
assert_eq!(proto, b"/proto2");
Expand Down