Skip to content

Commit 0f40e51

Browse files
thomaseizingerthomaseizinger
authored
core/muxing: Replace Into<io::Error> bound on StreamMuxer with std::error::Error (#2710)
* core/muxing: Remove `Into<io::Error>` bound from `StreamMuxer::Error` This allows us to preserve the type information of a muxer's concrete error as long as possible. For `StreamMuxerBox`, we leverage `io::Error`'s capability of wrapping any error that implements `Into<Box<dyn Error>>`. * Use `?` in `Connection::poll` * Use `?` in `muxing::boxed::Wrap` * Use `futures::ready!` in `muxing::boxed::Wrap` * Fill PR number into changelog * Put `Error + Send + Sync` bounds directly on `StreamMuxer::Error` * Move `Send + Sync` bounds to higher layers * Use `map_inbound_stream` helper * Update changelog to match new implementation
1 parent eb490c0 commit 0f40e51

File tree

8 files changed

+49
-45
lines changed

8 files changed

+49
-45
lines changed

‎core/CHANGELOG.md‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3,9 +3,11 @@
33
- Introduce `StreamMuxerEvent::map_inbound_stream`. See [PR 2691].
44
- Remove `{read,write,flush,shutdown,destroy}_substream` functions from `StreamMuxer` trait
55
in favor of forcing `StreamMuxer::Substream` to implement `AsyncRead + AsyncWrite`. See [PR 2707].
6+
- Replace `Into<std::io::Error>` bound on `StreamMuxer::Error` with `std::error::Error`. See [PR 2710].
67

78
[PR 2691]: https://github.com/libp2p/rust-libp2p/pull/2691
89
[PR 2707]: https://github.com/libp2p/rust-libp2p/pull/2707
10+
[PR 2710]: https://github.com/libp2p/rust-libp2p/pull/2710
911

1012
# 0.33.0
1113

‎core/src/either.rs‎

Lines changed: 7 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -203,7 +203,7 @@ where
203203
{
204204
type Substream = EitherOutput<A::Substream, B::Substream>;
205205
type OutboundSubstream = EitherOutbound<A, B>;
206-
type Error = io::Error;
206+
type Error = EitherError<A::Error, B::Error>;
207207

208208
fn poll_event(
209209
&self,
@@ -212,11 +212,11 @@ where
212212
match self {
213213
EitherOutput::First(inner) => inner
214214
.poll_event(cx)
215-
.map_err(|e| e.into())
215+
.map_err(EitherError::A)
216216
.map_ok(|event| event.map_inbound_stream(EitherOutput::First)),
217217
EitherOutput::Second(inner) => inner
218218
.poll_event(cx)
219-
.map_err(|e| e.into())
219+
.map_err(EitherError::B)
220220
.map_ok(|event| event.map_inbound_stream(EitherOutput::Second)),
221221
}
222222
}
@@ -237,11 +237,11 @@ where
237237
(EitherOutput::First(ref inner), EitherOutbound::A(ref mut substream)) => inner
238238
.poll_outbound(cx, substream)
239239
.map(|p| p.map(EitherOutput::First))
240-
.map_err(|e| e.into()),
240+
.map_err(EitherError::A),
241241
(EitherOutput::Second(ref inner), EitherOutbound::B(ref mut substream)) => inner
242242
.poll_outbound(cx, substream)
243243
.map(|p| p.map(EitherOutput::Second))
244-
.map_err(|e| e.into()),
244+
.map_err(EitherError::B),
245245
_ => panic!("Wrong API usage"),
246246
}
247247
}
@@ -261,8 +261,8 @@ where
261261

262262
fn poll_close(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
263263
match self {
264-
EitherOutput::First(inner) => inner.poll_close(cx).map_err(|e| e.into()),
265-
EitherOutput::Second(inner) => inner.poll_close(cx).map_err(|e| e.into()),
264+
EitherOutput::First(inner) => inner.poll_close(cx).map_err(EitherError::A),
265+
EitherOutput::Second(inner) => inner.poll_close(cx).map_err(EitherError::B),
266266
}
267267
}
268268
}

‎core/src/muxing.rs‎

Lines changed: 1 addition & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -52,7 +52,6 @@
5252
5353
use futures::{task::Context, task::Poll, AsyncRead, AsyncWrite};
5454
use multiaddr::Multiaddr;
55-
use std::io;
5655

5756
pub use self::boxed::StreamMuxerBox;
5857
pub use self::boxed::SubstreamBox;
@@ -76,7 +75,7 @@ pub trait StreamMuxer {
7675
type OutboundSubstream;
7776

7877
/// Error type of the muxer
79-
type Error: Into<io::Error>;
78+
type Error: std::error::Error;
8079

8180
/// Polls for a connection-wide event.
8281
///

‎core/src/muxing/boxed.rs‎

Lines changed: 18 additions & 20 deletions
Original file line numberDiff line numberDiff line change
@@ -1,8 +1,9 @@
11
use crate::muxing::StreamMuxerEvent;
22
use crate::StreamMuxer;
33
use fnv::FnvHashMap;
4-
use futures::{AsyncRead, AsyncWrite};
4+
use futures::{ready, AsyncRead, AsyncWrite};
55
use parking_lot::Mutex;
6+
use std::error::Error;
67
use std::fmt;
78
use std::io;
89
use std::io::{IoSlice, IoSliceMut};
@@ -38,6 +39,7 @@ impl<T> StreamMuxer for Wrap<T>
3839
where
3940
T: StreamMuxer,
4041
T::Substream: Send + Unpin + 'static,
42+
T::Error: Send + Sync + 'static,
4143
{
4244
type Substream = SubstreamBox;
4345
type OutboundSubstream = usize; // TODO: use a newtype
@@ -48,18 +50,10 @@ where
4850
&self,
4951
cx: &mut Context<'_>,
5052
) -> Poll<Result<StreamMuxerEvent<Self::Substream>, Self::Error>> {
51-
let substream = match self.inner.poll_event(cx) {
52-
Poll::Pending => return Poll::Pending,
53-
Poll::Ready(Ok(StreamMuxerEvent::AddressChange(a))) => {
54-
return Poll::Ready(Ok(StreamMuxerEvent::AddressChange(a)))
55-
}
56-
Poll::Ready(Ok(StreamMuxerEvent::InboundSubstream(s))) => s,
57-
Poll::Ready(Err(err)) => return Poll::Ready(Err(err.into())),
58-
};
53+
let event = ready!(self.inner.poll_event(cx).map_err(into_io_error)?)
54+
.map_inbound_stream(SubstreamBox::new);
5955

60-
Poll::Ready(Ok(StreamMuxerEvent::InboundSubstream(SubstreamBox::new(
61-
substream,
62-
))))
56+
Poll::Ready(Ok(event))
6357
}
6458

6559
#[inline]
@@ -77,16 +71,12 @@ where
7771
substream: &mut Self::OutboundSubstream,
7872
) -> Poll<Result<Self::Substream, Self::Error>> {
7973
let mut list = self.outbound.lock();
80-
let substream = match self
74+
let stream = ready!(self
8175
.inner
8276
.poll_outbound(cx, list.get_mut(substream).unwrap())
83-
{
84-
Poll::Pending => return Poll::Pending,
85-
Poll::Ready(Ok(s)) => s,
86-
Poll::Ready(Err(err)) => return Poll::Ready(Err(err.into())),
87-
};
77+
.map_err(into_io_error)?);
8878

89-
Poll::Ready(Ok(SubstreamBox::new(substream)))
79+
Poll::Ready(Ok(SubstreamBox::new(stream)))
9080
}
9181

9282
#[inline]
@@ -98,17 +88,25 @@ where
9888

9989
#[inline]
10090
fn poll_close(&self, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
101-
self.inner.poll_close(cx).map_err(|e| e.into())
91+
self.inner.poll_close(cx).map_err(into_io_error)
10292
}
10393
}
10494

95+
fn into_io_error<E>(err: E) -> io::Error
96+
where
97+
E: Error + Send + Sync + 'static,
98+
{
99+
io::Error::new(io::ErrorKind::Other, err)
100+
}
101+
105102
impl StreamMuxerBox {
106103
/// Turns a stream muxer into a `StreamMuxerBox`.
107104
pub fn new<T>(muxer: T) -> StreamMuxerBox
108105
where
109106
T: StreamMuxer + Send + Sync + 'static,
110107
T::OutboundSubstream: Send,
111108
T::Substream: Send + Unpin + 'static,
109+
T::Error: Send + Sync + 'static,
112110
{
113111
let wrap = Wrap {
114112
inner: muxer,

‎core/src/transport/upgrade.rs‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -302,6 +302,7 @@ impl<T> Multiplexed<T> {
302302
M: StreamMuxer + Send + Sync + 'static,
303303
M::Substream: Send + Unpin + 'static,
304304
M::OutboundSubstream: Send + 'static,
305+
M::Error: Send + Sync + 'static,
305306
{
306307
boxed(self.map(|(i, m), _| (i, StreamMuxerBox::new(m))))
307308
}

‎swarm/src/connection.rs‎

Lines changed: 8 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -137,40 +137,38 @@ where
137137
) -> Poll<Result<Event<THandler::OutEvent>, ConnectionError<THandler::Error>>> {
138138
loop {
139139
// Poll the handler for new events.
140-
match self.handler.poll(cx) {
140+
match self.handler.poll(cx)? {
141141
Poll::Pending => {}
142-
Poll::Ready(Ok(handler_wrapper::Event::OutboundSubstreamRequest(user_data))) => {
142+
Poll::Ready(handler_wrapper::Event::OutboundSubstreamRequest(user_data)) => {
143143
self.muxing.open_substream(user_data);
144144
continue;
145145
}
146-
Poll::Ready(Ok(handler_wrapper::Event::Custom(event))) => {
146+
Poll::Ready(handler_wrapper::Event::Custom(event)) => {
147147
return Poll::Ready(Ok(Event::Handler(event)));
148148
}
149-
Poll::Ready(Err(err)) => return Poll::Ready(Err(err.into())),
150149
}
151150

152151
// Perform I/O on the connection through the muxer, informing the handler
153152
// of new substreams.
154-
match self.muxing.poll(cx) {
153+
match self.muxing.poll(cx)? {
155154
Poll::Pending => {}
156-
Poll::Ready(Ok(SubstreamEvent::InboundSubstream { substream })) => {
155+
Poll::Ready(SubstreamEvent::InboundSubstream { substream }) => {
157156
self.handler
158157
.inject_substream(substream, SubstreamEndpoint::Listener);
159158
continue;
160159
}
161-
Poll::Ready(Ok(SubstreamEvent::OutboundSubstream {
160+
Poll::Ready(SubstreamEvent::OutboundSubstream {
162161
user_data,
163162
substream,
164-
})) => {
163+
}) => {
165164
let endpoint = SubstreamEndpoint::Dialer(user_data);
166165
self.handler.inject_substream(substream, endpoint);
167166
continue;
168167
}
169-
Poll::Ready(Ok(SubstreamEvent::AddressChange(address))) => {
168+
Poll::Ready(SubstreamEvent::AddressChange(address)) => {
170169
self.handler.inject_address_change(&address);
171170
return Poll::Ready(Ok(Event::AddressChange(address)));
172171
}
173-
Poll::Ready(Err(err)) => return Poll::Ready(Err(ConnectionError::IO(err))),
174172
}
175173

176174
return Poll::Pending;

‎swarm/src/connection/error.rs‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,12 @@ impl<THandlerErr> From<handler_wrapper::Error<THandlerErr>> for ConnectionError<
7575
}
7676
}
7777

78+
impl<THandlerErr> From<io::Error> for ConnectionError<THandlerErr> {
79+
fn from(error: io::Error) -> Self {
80+
ConnectionError::IO(error)
81+
}
82+
}
83+
7884
/// Errors that can occur in the context of a pending outgoing `Connection`.
7985
///
8086
/// Note: Addresses for an outbound connection are dialed in parallel. Thus, compared to

‎swarm/src/connection/substream.rs‎

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -23,7 +23,7 @@ use libp2p_core::multiaddr::Multiaddr;
2323
use libp2p_core::muxing::{StreamMuxer, StreamMuxerEvent};
2424
use smallvec::SmallVec;
2525
use std::sync::Arc;
26-
use std::{fmt, io::Error as IoError, pin::Pin, task::Context, task::Poll};
26+
use std::{fmt, pin::Pin, task::Context, task::Poll};
2727

2828
/// Endpoint for a received substream.
2929
#[derive(Debug, Copy, Clone, PartialEq, Eq)]
@@ -134,7 +134,7 @@ where
134134
pub fn poll(
135135
&mut self,
136136
cx: &mut Context<'_>,
137-
) -> Poll<Result<SubstreamEvent<TMuxer, TUserData>, IoError>> {
137+
) -> Poll<Result<SubstreamEvent<TMuxer, TUserData>, TMuxer::Error>> {
138138
// Polling inbound substream.
139139
match self.inner.poll_event(cx) {
140140
Poll::Ready(Ok(StreamMuxerEvent::InboundSubstream(substream))) => {
@@ -143,7 +143,7 @@ where
143143
Poll::Ready(Ok(StreamMuxerEvent::AddressChange(addr))) => {
144144
return Poll::Ready(Ok(SubstreamEvent::AddressChange(addr)))
145145
}
146-
Poll::Ready(Err(err)) => return Poll::Ready(Err(err.into())),
146+
Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
147147
Poll::Pending => {}
148148
}
149149

@@ -164,7 +164,7 @@ where
164164
}
165165
Poll::Ready(Err(err)) => {
166166
self.inner.destroy_outbound(outbound);
167-
return Poll::Ready(Err(err.into()));
167+
return Poll::Ready(Err(err));
168168
}
169169
}
170170
}
@@ -203,13 +203,13 @@ impl<TMuxer> Future for Close<TMuxer>
203203
where
204204
TMuxer: StreamMuxer,
205205
{
206-
type Output = Result<(), IoError>;
206+
type Output = Result<(), TMuxer::Error>;
207207

208208
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
209209
match self.muxer.poll_close(cx) {
210210
Poll::Pending => Poll::Pending,
211211
Poll::Ready(Ok(())) => Poll::Ready(Ok(())),
212-
Poll::Ready(Err(err)) => Poll::Ready(Err(err.into())),
212+
Poll::Ready(Err(err)) => Poll::Ready(Err(err)),
213213
}
214214
}
215215
}

0 commit comments

Comments
 (0)