Drop dep tokio's io-util feat as it broke MSRV and isn't useful

We use `tokio`'s `io-util` feature to provide the
`Async{Read,Write}Ext` traits, which allow us to simply launch a
read future or `poll_write` directly as well as `split` the
`TcpStream` into a read/write half. However, these traits aren't
actually doing much for us - they are really just wrapping the
`readable` future (which we can trivially use ourselves) and
`poll_write` isn't doing anything for us that `poll_write_ready`
can't.

Similarly, the split logic is actually just `Arc`ing the
`TcpStream` and busy-waiting when an operation is busy to prevent
concurrent reads/writes. However, there's no reason to prevent
concurrent access at the stream level - we aren't ever concurrently
writing or reading (though we may concurrently read and write,
which is fine).

Worse, the `io-util` feature broke MSRV (though they're likely to
fix this upstream) and carries two additional dependencies (only
one on the latest upstream tokio).

Thus, we simply drop the dependency here.

Fixes lightningdevkit#2527.

Author: TheBlueMatt

lightning-net-tokio/Cargo.toml

@@ -17,8 +17,8 @@ rustdoc-args = ["--cfg", "docsrs"]
 [dependencies]
 bitcoin = "0.29.0"
 lightning = { version = "0.0.116", path = "../lightning" }
-tokio = { version = "1.0", features = [ "io-util", "rt", "sync", "net", "time" ] }
+tokio = { version = "1.0", features = [ "rt", "sync", "net", "time" ] }
 
 [dev-dependencies]
-tokio = { version = "1.14", features = [ "io-util", "macros", "rt", "rt-multi-thread", "sync", "net", "time" ] }
+tokio = { version = "1.14", features = [ "macros", "rt", "rt-multi-thread", "sync", "net", "time" ] }
 lightning = { version = "0.0.116", path = "../lightning", features = ["_test_utils"] }

lightning-net-tokio/src/lib.rs

@@ -31,10 +31,10 @@
 
 use bitcoin::secp256k1::PublicKey;
 
-use tokio::net::TcpStream;
+use tokio::net::{tcp, TcpStream};
 use tokio::{io, time};
 use tokio::sync::mpsc;
-use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt};
+use tokio::io::AsyncWrite;
 
 use lightning::ln::peer_handler;
 use lightning::ln::peer_handler::SocketDescriptor as LnSocketTrait;
@@ -59,7 +59,7 @@ static ID_COUNTER: AtomicU64 = AtomicU64::new(0);
 // define a trivial two- and three- select macro with the specific types we need and just use that.
 
 pub(crate) enum SelectorOutput {
-	A(Option<()>), B(Option<()>), C(tokio::io::Result<usize>),
+	A(Option<()>), B(Option<()>), C(tokio::io::Result<()>),
 }
 
 pub(crate) struct TwoSelector<
@@ -87,15 +87,15 @@ impl<
 }
 
 pub(crate) struct ThreeSelector<
-	A: Future<Output=Option<()>> + Unpin, B: Future<Output=Option<()>> + Unpin, C: Future<Output=tokio::io::Result<usize>> + Unpin
+	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,
 }
 
 impl<
-	A: Future<Output=Option<()>> + Unpin, B: Future<Output=Option<()>> + Unpin, C: Future<Output=tokio::io::Result<usize>> + Unpin
+	A: Future<Output=Option<()>> + Unpin, B: Future<Output=Option<()>> + Unpin, C: Future<Output=tokio::io::Result<()>> + Unpin
 > Future for ThreeSelector<A, B, C> {
 	type Output = SelectorOutput;
 	fn poll(mut self: Pin<&mut Self>, ctx: &mut task::Context<'_>) -> Poll<SelectorOutput> {
@@ -119,7 +119,7 @@ impl<
 /// Connection object (in an Arc<Mutex<>>) in each SocketDescriptor we create as well as in the
 /// read future (which is returned by schedule_read).
 struct Connection {
-	writer: Option<io::WriteHalf<TcpStream>>,
+	writer: Option<Arc<TcpStream>>,
 	// Because our PeerManager is templated by user-provided types, and we can't (as far as I can
 	// tell) have a const RawWakerVTable built out of templated functions, we need some indirection
 	// between being woken up with write-ready and calling PeerManager::write_buffer_space_avail.
@@ -156,7 +156,7 @@ impl Connection {
 	async fn schedule_read<PM: Deref + 'static + Send + Sync + Clone>(
 		peer_manager: PM,
 		us: Arc<Mutex<Self>>,
-		mut reader: io::ReadHalf<TcpStream>,
+		reader: Arc<TcpStream>,
 		mut read_wake_receiver: mpsc::Receiver<()>,
 		mut write_avail_receiver: mpsc::Receiver<()>,
 	) where PM::Target: APeerManager<Descriptor = SocketDescriptor> {
@@ -200,7 +200,7 @@ impl Connection {
 				ThreeSelector {
 					a: Box::pin(write_avail_receiver.recv()),
 					b: Box::pin(read_wake_receiver.recv()),
-					c: Box::pin(reader.read(&mut buf)),
+					c: Box::pin(reader.readable()),
 				}.await
 			};
 			match select_result {
@@ -211,8 +211,9 @@ impl Connection {
 					}
 				},
 				SelectorOutput::B(_) => {},
-				SelectorOutput::C(read) => {
-					match read {
+				SelectorOutput::C(res) => {
+					if res.is_err() { break Disconnect::PeerDisconnected; }
+					match reader.try_read(&mut buf) {
 						Ok(0) => break Disconnect::PeerDisconnected,
 						Ok(len) => {
 							let read_res = peer_manager.as_ref().read_event(&mut our_descriptor, &buf[0..len]);
@@ -226,7 +227,11 @@ impl Connection {
 								Err(_) => break Disconnect::CloseConnection,
 							}
 						},
-						Err(_) => break Disconnect::PeerDisconnected,
+						Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
+							// readable() is allowed to spuriously wake, so we have to handle
+							// WouldBlock here.
+						},
+						Err(e) => break Disconnect::PeerDisconnected,
 					}
 				},
 			}
@@ -239,18 +244,14 @@ impl Connection {
 			// here.
 			let _ = tokio::task::yield_now().await;
 		};
-		let writer_option = us.lock().unwrap().writer.take();
-		if let Some(mut writer) = writer_option {
-			// If the socket is already closed, shutdown() will fail, so just ignore it.
-			let _ = writer.shutdown().await;
-		}
+		us.lock().unwrap().writer.take();
 		if let Disconnect::PeerDisconnected = disconnect_type {
 			peer_manager.as_ref().socket_disconnected(&our_descriptor);
 			peer_manager.as_ref().process_events();
 		}
 	}
 
-	fn new(stream: StdTcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
+	fn new(stream: StdTcpStream) -> (Arc<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
 		// We only ever need a channel of depth 1 here: if we returned a non-full write to the
 		// PeerManager, we will eventually get notified that there is room in the socket to write
 		// new bytes, which will generate an event. That event will be popped off the queue before
@@ -262,11 +263,11 @@ impl Connection {
 		// false.
 		let (read_waker, read_receiver) = mpsc::channel(1);
 		stream.set_nonblocking(true).unwrap();
-		let (reader, writer) = io::split(TcpStream::from_std(stream).unwrap());
+		let tokio_stream = Arc::new(TcpStream::from_std(stream).unwrap());
 
-		(reader, write_receiver, read_receiver,
+		(Arc::clone(&tokio_stream), write_receiver, read_receiver,
 		Arc::new(Mutex::new(Self {
-			writer: Some(writer), write_avail, read_waker, read_paused: false,
+			writer: Some(tokio_stream), write_avail, read_waker, read_paused: false,
 			rl_requested_disconnect: false,
 			id: ID_COUNTER.fetch_add(1, Ordering::AcqRel)
 		})))
@@ -462,9 +463,9 @@ impl SocketDescriptor {
 }
 impl peer_handler::SocketDescriptor for SocketDescriptor {
 	fn send_data(&mut self, data: &[u8], resume_read: bool) -> usize {
-		// To send data, we take a lock on our Connection to access the WriteHalf of the TcpStream,
-		// writing to it if there's room in the kernel buffer, or otherwise create a new Waker with
-		// a SocketDescriptor in it which can wake up the write_avail Sender, waking up the
+		// To send data, we take a lock on our Connection to access the TcpStream, writing to it if
+		// there's room in the kernel buffer, or otherwise create a new Waker with a
+		// SocketDescriptor in it which can wake up the write_avail Sender, waking up the
 		// processing future which will call write_buffer_space_avail and we'll end up back here.
 		let mut us = self.conn.lock().unwrap();
 		if us.writer.is_none() {
@@ -484,24 +485,18 @@ impl peer_handler::SocketDescriptor for SocketDescriptor {
 		let mut ctx = task::Context::from_waker(&waker);
 		let mut written_len = 0;
 		loop {
-			match std::pin::Pin::new(us.writer.as_mut().unwrap()).poll_write(&mut ctx, &data[written_len..]) {
-				task::Poll::Ready(Ok(res)) => {
-					// The tokio docs *seem* to indicate this can't happen, and I certainly don't
-					// know how to handle it if it does (cause it should be a Poll::Pending
-					// instead):
-					assert_ne!(res, 0);
-					written_len += res;
-					if written_len == data.len() { return written_len; }
-				},
-				task::Poll::Ready(Err(e)) => {
-					// The tokio docs *seem* to indicate this can't happen, and I certainly don't
-					// know how to handle it if it does (cause it should be a Poll::Pending
-					// instead):
-					assert_ne!(e.kind(), io::ErrorKind::WouldBlock);
-					// Probably we've already been closed, just return what we have and let the
-					// read thread handle closing logic.
-					return written_len;
+			match us.writer.as_ref().unwrap().poll_write_ready(&mut ctx) {
+				task::Poll::Ready(Ok(())) => {
+					match us.writer.as_ref().unwrap().try_write(&data[written_len..]) {
+						Ok(res) => {
+							debug_assert_ne!(res, 0);
+							written_len += res;
+							if written_len == data.len() { return written_len; }
+						},
+						Err(e) => return written_len,
+					}
 				},
+				task::Poll::Ready(Err(e)) => return written_len,
 				task::Poll::Pending => {
 					// We're queued up for a write event now, but we need to make sure we also
 					// pause read given we're now waiting on the remote end to ACK (and in