Avoid calling PeerManager::process_events more often than needed

Instead of calling `process_events` after each byte read from a
peer, only call it when we have no data left to process. This
should reduce write lock contention somewhat inside `PeerManager`.

Author: TheBlueMatt

lightning-net-tokio/src/lib.rs

@@ -71,10 +71,10 @@
 
 use bitcoin::secp256k1::key::PublicKey;
 
-use tokio::net::TcpStream;
+use tokio::net::{TcpStream, tcp::OwnedReadHalf, tcp::OwnedWriteHalf};
 use tokio::{io, time};
 use tokio::sync::mpsc;
-use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt};
+use tokio::io::{AsyncWrite, AsyncWriteExt};
 
 use lightning::ln::peer_handler;
 use lightning::ln::peer_handler::SocketDescriptor as LnSocketTrait;
@@ -96,7 +96,7 @@ static ID_COUNTER: AtomicU64 = AtomicU64::new(0);
 /// 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<OwnedWriteHalf>,
 	// 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.
@@ -118,7 +118,7 @@ struct Connection {
 	id: u64,
 }
 impl Connection {
-	async fn schedule_read<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>, Arc<UMH>>>, us: Arc<Mutex<Self>>, mut reader: io::ReadHalf<TcpStream>, mut read_wake_receiver: mpsc::Receiver<()>, mut write_avail_receiver: mpsc::Receiver<()>) where
+	async fn schedule_read<CMH, RMH, L, UMH>(peer_manager: Arc<peer_handler::PeerManager<SocketDescriptor, Arc<CMH>, Arc<RMH>, Arc<L>, Arc<UMH>>>, us: Arc<Mutex<Self>>, reader: OwnedReadHalf, mut read_wake_receiver: mpsc::Receiver<()>, mut write_avail_receiver: mpsc::Receiver<()>) where
 			CMH: ChannelMessageHandler + 'static,
 			RMH: RoutingMessageHandler + 'static,
 			L: Logger + 'static + ?Sized,
@@ -140,7 +140,7 @@ impl Connection {
 			// Rust-Lightning that the socket is gone.
 			PeerDisconnected
 		}
-		let disconnect_type = loop {
+		let disconnect_type = 'socket_loop: loop {
 			let read_paused = {
 				let us_lock = us.lock().unwrap();
 				if us_lock.rl_requested_disconnect {
@@ -156,21 +156,33 @@ impl Connection {
 					}
 				},
 				_ = read_wake_receiver.recv() => {},
-				read = reader.read(&mut buf), if !read_paused => match read {
-					Ok(0) => break Disconnect::PeerDisconnected,
-					Ok(len) => {
-						let read_res = peer_manager.read_event(&mut our_descriptor, &buf[0..len]);
-						let mut us_lock = us.lock().unwrap();
-						match read_res {
-							Ok(pause_read) => {
-								if pause_read {
-									us_lock.read_paused = true;
+				_ = reader.as_ref().readable(), if !read_paused => {
+					// We fully drain the socket before we move on and call `process_events` or
+					// possibly call `write_buffer_space_avail`. LDK will ensure we aren't waiting
+					// for write and overfill our write buffer by telling us to pause read, which
+					// will break the loop.
+					// This prevents us from calling `process_events` more often than needed while
+					// there's still work to be done processing incoming data.
+					'read_loop: loop {
+						match reader.as_ref().try_read(&mut buf) {
+							Ok(0) => break 'socket_loop Disconnect::PeerDisconnected,
+							Ok(len) => {
+								let read_res = peer_manager.read_event(&mut our_descriptor, &buf[0..len]);
+								let mut us_lock = us.lock().unwrap();
+								match read_res {
+									Ok(pause_read) => {
+										if pause_read {
+											us_lock.read_paused = true;
+											break 'read_loop;
+										}
+									},
+									Err(_) => break 'socket_loop Disconnect::CloseConnection,
 								}
 							},
-							Err(_) => break Disconnect::CloseConnection,
+							Err(e) if e.kind() == io::ErrorKind::WouldBlock => break 'read_loop,
+							Err(_) => break 'socket_loop Disconnect::PeerDisconnected,
 						}
-					},
-					Err(_) => break Disconnect::PeerDisconnected,
+					}
 				},
 			}
 			peer_manager.process_events();
@@ -186,7 +198,7 @@ impl Connection {
 		}
 	}
 
-	fn new(stream: StdTcpStream) -> (io::ReadHalf<TcpStream>, mpsc::Receiver<()>, mpsc::Receiver<()>, Arc<Mutex<Self>>) {
+	fn new(stream: StdTcpStream) -> (OwnedReadHalf, 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
@@ -198,7 +210,7 @@ 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 (reader, writer) = TcpStream::from_std(stream).unwrap().into_split();
 
 		(reader, write_receiver, read_receiver,
 		Arc::new(Mutex::new(Self {