fluencelabs/rust-libp2p
1
0
Fork 0
mirror of https://github.com/fluencelabs/rust-libp2p synced 2025-05-30 19:21:19 +00:00
Code Issues Projects Releases Wiki Activity

Remove even substream numbers from mplex (#869)

Browse Source 
* Remove even substream numbers from mplex

* Add panic on overflow
This commit is contained in:
Pierre Krieger 2019-01-25 11:26:37 +01:00 committed by GitHub
parent d9b30c6c6a
commit f492cb3786
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 7 additions and 6 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
muxers/mplex/src/lib.rs
View File

@ -95,7 +95,7 @@ impl MplexConfig {
} }
#[inline] #[inline]
fn upgrade<C>(self, i: C, endpoint: Endpoint) -> Multiplex<C> fn upgrade<C>(self, i: C) -> Multiplex<C>
where where
C: AsyncRead + AsyncWrite C: AsyncRead + AsyncWrite
{ {
@ -107,7 +107,7 @@ impl MplexConfig {
config: self, config: self,
buffer: Vec::with_capacity(cmp::min(max_buffer_len, 512)), buffer: Vec::with_capacity(cmp::min(max_buffer_len, 512)),
opened_substreams: Default::default(), opened_substreams: Default::default(),
next_outbound_stream_id: if endpoint == Endpoint::Dialer { 0 } else { 1 }, next_outbound_stream_id: 0,
notifier_read: Arc::new(Notifier { notifier_read: Arc::new(Notifier {
to_notify: Mutex::new(Default::default()), to_notify: Mutex::new(Default::default()),
}), }),
@ -163,7 +163,7 @@ where
type Future = future::FutureResult<Self::Output, IoError>; type Future = future::FutureResult<Self::Output, IoError>;
fn upgrade_inbound(self, socket: C, _: Self::Info) -> Self::Future { fn upgrade_inbound(self, socket: C, _: Self::Info) -> Self::Future {
future::ok(self.upgrade(socket, Endpoint::Listener)) future::ok(self.upgrade(socket))
} }
} }
@ -176,7 +176,7 @@ where
type Future = future::FutureResult<Self::Output, IoError>; type Future = future::FutureResult<Self::Output, IoError>;
fn upgrade_outbound(self, socket: C, _: Self::Info) -> Self::Future { fn upgrade_outbound(self, socket: C, _: Self::Info) -> Self::Future {
future::ok(self.upgrade(socket, Endpoint::Dialer)) future::ok(self.upgrade(socket))
} }
} }
@ -200,7 +200,7 @@ struct MultiplexInner<C> {
// The `Endpoint` value denotes who initiated the substream from our point of view // The `Endpoint` value denotes who initiated the substream from our point of view
// (see note [StreamId]). // (see note [StreamId]).
opened_substreams: FnvHashSet<(u32, Endpoint)>, opened_substreams: FnvHashSet<(u32, Endpoint)>,
// Id of the next outgoing substream. Should always increase by two. // Id of the next outgoing substream.
next_outbound_stream_id: u32, next_outbound_stream_id: u32,
/// List of tasks to notify when a read event happens on the underlying stream. /// List of tasks to notify when a read event happens on the underlying stream.
notifier_read: Arc<Notifier>, notifier_read: Arc<Notifier>,
@ -382,7 +382,8 @@ where C: AsyncRead + AsyncWrite
// Assign a substream ID now. // Assign a substream ID now.
let substream_id = { let substream_id = {
let n = inner.next_outbound_stream_id; let n = inner.next_outbound_stream_id;
inner.next_outbound_stream_id += 2; inner.next_outbound_stream_id = inner.next_outbound_stream_id.checked_add(1)
.expect("Mplex substream ID overflowed");
n n
}; };