Rewrite ConnectionReuse #355

tomaka · 2018-07-22T16:18:30Z

Second rewrite. This time with more manual polling, otherwise we have tons of Arcs.

The PR is not finished as the code should be made cleaner (more comments), and tests should be written.

tomaka · 2018-07-23T14:16:57Z

Note for self: the latest commit (fix dialer not counting as active substream) is not included in polkadot-tmp because of #361.
Since I want to keep polkadot-tmp more or less in sync with master, care must be taken when merging this.

…to connec-reuse-rewrite

gnunicorn

Thanks for that!

Reading through it though, things feel a little all over the place (still) and with replicated code parts, especially around the important connection management and upgrade parts, making this prone to easily breaking it with future changes (or forgetting to copy other parts).

May I suggest, rather than using shared: Arc<Mutex<Shared<T, C>>> and having each user deal with it (hopefully properly), to encapsulate it within a Wrapper and centralize the repeated codebase there?

Something like

 struct ConnectionManager {
   inner: Arc<Mutex<Shared<T, C>>>
}

impl ConnectionManager {
    pub fn remove_substream(connec_id: u64, addr: &Multiaddr)
    pub fn poll_incoming(listener: Option<u64>)
    pub fn insert_connection( ....
}

gnunicorn · 2018-07-30T12:15:41Z

core/src/connection_reuse.rs

+    transport: UpgradedNode<T, C>,
+
+    /// All the connections that were opened, whether successful and/or active or not.
+    // TODO: this will grow forever


well, maybe use an LRU-Cache instead to prevent boundless growth? Doesn't have to be addressed in this PR though we might want to put this TODO into a ticket otherwise (could be nice for new contributors)...

gnunicorn · 2018-07-31T10:16:08Z

core/src/connection_reuse.rs

+        muxer: M,
+        /// Next incoming substream.
+        next_incoming: M::InboundSubstream,
+        /// Future of the address of the client.


~~Future of~~

gnunicorn · 2018-07-31T10:19:03Z

core/src/connection_reuse.rs

+
+    /// The `PeerState` is poisonned. Happens if a panic happened while executing some of the
+    /// functions.
+    Poisonned,


Typo: Poison n ed

gnunicorn · 2018-07-31T10:38:35Z

core/src/connection_reuse.rs

+                },
+            };
+
+            match mem::replace(&mut *connec, PeerState::Poisonned) {


Why are we replacing a potentially just created Pending-State with a Poisoned one immediately and then replace it within every match, too? Why not do a regular match?

You can't switch from Pending to Active if you use a regular match block.

gnunicorn · 2018-07-31T10:58:14Z

core/src/connection_reuse.rs

+    type Error = IoError;
+
+    fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
+        loop {


this loop is over 110 lines long and contains a lot of complexity, mem::replace (which we immediately overwrite), interdependence and early returns - requiring a huge mental model to keep in your head to follow all that is happening. Any chance we could split this up into easier consumable chunks?

gnunicorn · 2018-07-31T11:14:38Z

core/src/connection_reuse.rs

+                Ok(Async::Ready(substream))
+            },
+            Ok(Async::Ready(None)) | Ok(Async::NotReady) => {
+                // TODO: will add an element to the list every time


comment not true with hashmap anymore.

gnunicorn · 2018-07-31T11:15:08Z

core/src/connection_reuse.rs

+
+    for (addr, state) in shared.connections.iter_mut() {
+        match *state {
+            PeerState::Active { ref mut next_incoming, ref muxer, ref mut num_substreams, connection_id, ref client_addr, listener_id } => {


please wrap.

gnunicorn · 2018-07-31T11:25:03Z

core/src/connection_reuse.rs

+///
+/// Returns `Ready(None)` if no connection is matching the `listener`. Returns `NotReady` if
+/// one or more connections are matching the `listener` but they are not ready.
+fn poll_incoming<T, C>(shared_arc: &Arc<Mutex<Shared<T, C>>>, shared: &mut Shared<T, C>, listener: Option<u64>)


having both shared_arc: &Arc<Mutex<Shared<T, C>>> and shared: &mut Shared<T, C> as parameters feels very odd, and makes me wonder if this isn't supposed to be somewhere else closed to Shared itself.

gnunicorn · 2018-07-31T11:26:58Z

core/src/connection_reuse.rs

+}
+
+/// Removes one substream from an active connection. Closes the connection if necessary.
+fn remove_one_substream<T, C>(shared: &mut Shared<T, C>, connec_id: u64, addr: &Multiaddr)


this should be closer to Shared as well.

gnunicorn · 2018-07-31T11:28:40Z

core/tests/multiplex.rs

-use libp2p_tcp_transport::TcpConfig;
-use std::sync::{atomic, mpsc};
+use libp2p_core::{Multiaddr, MuxedTransport, StreamMuxer, Transport, transport};
+use std::sync::atomic;


are you sure the changes in this file belong into this PR?

twittner · 2018-08-08T10:06:57Z

core/src/connection_reuse.rs

+                        Ok(Async::Ready(Some(inner))) => {
+                            trace!("New incoming substream from {}", client_addr);
+                            *num_substreams += 1;
+                            Ok((inner, connection_id.clone(), client_addr.clone()))
