Simplify net-tokio select logic

When `PeerManager` tells us to stop reading from a peer,
`lightning-net-tokio` will skip polling the TCP socket receiver.
This is great, but we had an entire separate struct to avoid
the second poll, which we remove here, replacing the missing future
with an `Option`.

Author: TheBlueMatt

lightning-net-tokio/src/lib.rs

@@ -61,43 +61,14 @@ pub(crate) enum SelectorOutput {
 	C(tokio::io::Result<()>),
 }
 
-pub(crate) struct TwoSelector<
-	A: Future<Output = Option<()>> + Unpin,
-	B: Future<Output = Option<()>> + Unpin,
-> {
-	pub a: A,
-	pub b: B,
-}
-
-impl<A: Future<Output = Option<()>> + Unpin, B: Future<Output = Option<()>> + Unpin> Future
-	for TwoSelector<A, B>
-{
-	type Output = SelectorOutput;
-	fn poll(mut self: Pin<&mut Self>, ctx: &mut task::Context<'_>) -> Poll<SelectorOutput> {
-		match Pin::new(&mut self.a).poll(ctx) {
-			Poll::Ready(res) => {
-				return Poll::Ready(SelectorOutput::A(res));
-			},
-			Poll::Pending => {},
-		}
-		match Pin::new(&mut self.b).poll(ctx) {
-			Poll::Ready(res) => {
-				return Poll::Ready(SelectorOutput::B(res));
-			},
-			Poll::Pending => {},
-		}
-		Poll::Pending
-	}
-}
-
 pub(crate) struct ThreeSelector<
 	A: Future<Output = Option<()>> + Unpin,
 	B: Future<Output = Option<()>> + Unpin,
 	C: Future<Output = tokio::io::Result<()>> + Unpin,
 > {
 	pub a: A,
 	pub b: B,
-	pub c: C,
+	pub c: Option<C>,
 }
 
 impl<
@@ -120,11 +91,13 @@ impl<
 			},
 			Poll::Pending => {},
 		}
-		match Pin::new(&mut self.c).poll(ctx) {
-			Poll::Ready(res) => {
-				return Poll::Ready(SelectorOutput::C(res));
-			},
-			Poll::Pending => {},
+		if let Some(c) = self.c.as_mut() {
+			match Pin::new(c).poll(ctx) {
+				Poll::Ready(res) => {
+					return Poll::Ready(SelectorOutput::C(res));
+				},
+				Poll::Pending => {},
+			}
 		}
 		Poll::Pending
 	}
@@ -198,29 +171,27 @@ impl Connection {
 			PeerDisconnected,
 		}
 		let disconnect_type = loop {
-			let read_paused = {
+			let selector = {
 				let us_lock = us.lock().unwrap();
 				if us_lock.rl_requested_disconnect {
 					break Disconnect::CloseConnection;
 				}
-				us_lock.read_paused
-			};
-			// TODO: Drop the Box'ing of the futures once Rust has pin-on-stack support.
-			let select_result = if read_paused {
-				TwoSelector {
-					a: Box::pin(write_avail_receiver.recv()),
-					b: Box::pin(read_wake_receiver.recv()),
-				}
-				.await
-			} else {
-				ThreeSelector {
-					a: Box::pin(write_avail_receiver.recv()),
-					b: Box::pin(read_wake_receiver.recv()),
-					c: Box::pin(reader.readable()),
+				if us_lock.read_paused {
+					// TODO: Drop the Box'ing of the futures once Rust has pin-on-stack support.
+					ThreeSelector {
+						a: Box::pin(write_avail_receiver.recv()),
+						b: Box::pin(read_wake_receiver.recv()),
+						c: None,
+					}
+				} else {
+					ThreeSelector {
+						a: Box::pin(write_avail_receiver.recv()),
+						b: Box::pin(read_wake_receiver.recv()),
+						c: Some(Box::pin(reader.readable())),
+					}
 				}
-				.await
 			};
-			match select_result {
+			match selector.await {
 				SelectorOutput::A(v) => {
 					assert!(v.is_some()); // We can't have dropped the sending end, its in the us Arc!
 					if peer_manager.as_ref().write_buffer_space_avail(&mut our_descriptor).is_err()