Change OutboundQueue flush api so that it flushes entire contents

The do_attempt_write_data() function would previously only flush
one item at a time. This would cause non-optimal batching in cases
where the queue was full and the RoutingMessageHandler was only asked
for a single item.

Instead, change the API to flush the entire contents of the queue.

Author: julianknutsen

lightning/src/ln/peers/handler.rs

@@ -81,10 +81,10 @@ pub(super) trait MessageQueuer {
 /// Trait representing a container that can try to flush data through a SocketDescriptor. Used by the
 /// PeerManager to handle flushing the outbound queue and flow control.
 pub(super) trait SocketDescriptorFlusher {
-	/// Write previously enqueued data to the SocketDescriptor. A return of false indicates the
-	/// underlying SocketDescriptor could not fulfill the send_data() call and the blocked state
-	/// has been set. Use unblock() when the SocketDescriptor may have more room.
-	fn try_flush_one(&mut self, descriptor: &mut impl SocketDescriptor) -> bool;
+	/// Attempts to write all previously enqueued data to the SocketDescriptor. A return of false
+	/// indicates the underlying SocketDescriptor could not fulfill the send_data() calls and the
+	/// blocked state has been set. Use unblock() when the SocketDescriptor may have more room.
+	fn try_flush(&mut self, descriptor: &mut impl SocketDescriptor) -> bool;
 
 	/// Clear the blocked state caused when a previous write failed
 	fn unblock(&mut self);
@@ -652,7 +652,7 @@ impl<Descriptor: SocketDescriptor, CM: Deref, RM: Deref, L: Deref, TransportImpl
 				break;
 			}
 
-			outbound_queue.try_flush_one(descriptor);
+			outbound_queue.try_flush(descriptor);
 		}
 	}
 

lightning/src/ln/peers/outbound_queue.rs

@@ -31,32 +31,37 @@ impl PayloadQueuer for OutboundQueue {
 }
 
 impl SocketDescriptorFlusher for OutboundQueue {
-	fn try_flush_one(&mut self, descriptor: &mut impl SocketDescriptor) -> bool {
+	fn try_flush(&mut self, descriptor: &mut impl SocketDescriptor) -> bool {
 		// Exit early if  a previous full write failed and haven't heard that there may be more
 		// room available
 		if self.blocked {
 			return false;
 		}
 
-		let full_write_succeeded = match self.buffer.front() {
-			None => true,
-			Some(next_buff) => {
-				let should_be_reading = self.buffer.len() < self.soft_limit;
-				let pending = &next_buff[self.buffer_first_msg_offset..];
-				let data_sent = descriptor.send_data(pending, should_be_reading);
-				self.buffer_first_msg_offset += data_sent;
-				self.buffer_first_msg_offset == next_buff.len()
+		loop {
+			if self.buffer.is_empty() {
+				return true;
 			}
-		};
 
-		if full_write_succeeded {
-			self.buffer_first_msg_offset = 0;
-			self.buffer.pop_front();
-		} else {
-			self.blocked = true;
+			let full_write_succeeded = match self.buffer.front() {
+				None => true,
+				Some(next_buff) => {
+					let should_be_reading = self.buffer.len() < self.soft_limit;
+					let pending = &next_buff[self.buffer_first_msg_offset..];
+					let data_sent = descriptor.send_data(pending, should_be_reading);
+					self.buffer_first_msg_offset += data_sent;
+					self.buffer_first_msg_offset == next_buff.len()
+				}
+			};
+
+			if full_write_succeeded {
+				self.buffer_first_msg_offset = 0;
+				self.buffer.pop_front();
+			} else {
+				self.blocked = true;
+				return false;
+			}
 		}
-
-		full_write_succeeded
 	}
 
 	fn unblock(&mut self) {
@@ -87,40 +92,40 @@ mod tests {
 	use super::*;
 	use ln::peers::test_util::*;
 
-	// Test that a try_flush_one() call with no queued data doesn't write anything
+	// Test that a try_flush() call with no queued data doesn't write anything
 	#[test]
 	fn empty_does_not_write() {
 		let mut descriptor = SocketDescriptorMock::new();
 		let mut empty = OutboundQueue::new(10);
 
-		assert!(empty.try_flush_one(&mut descriptor));
+		assert!(empty.try_flush(&mut descriptor));
 		descriptor.assert_called_with(vec![]);
 
 	}
 
-	// Test that try_flush_one() sends the push_back
+	// Test that try_flush() sends the push_back
 	#[test]
 	fn push_back_drain() {
 		let mut descriptor = SocketDescriptorMock::new();
 		let mut queue = OutboundQueue::new(10);
 
 		queue.push_back(vec![1]);
-		assert!(queue.try_flush_one(&mut descriptor));
+		assert!(queue.try_flush(&mut descriptor));
 
 		descriptor.assert_called_with(vec![(vec![1], true)]);
 	}
 
-	// Test that try_flush_one() sends just first push_back
+	// Test that try_flush() sends all
 	#[test]
 	fn push_back_push_back_drain_drain() {
 		let mut descriptor = SocketDescriptorMock::new();
 		let mut queue = OutboundQueue::new(10);
 
 		queue.push_back(vec![1]);
 		queue.push_back(vec![2]);
-		assert!(queue.try_flush_one(&mut descriptor));
+		assert!(queue.try_flush(&mut descriptor));
 
-		descriptor.assert_called_with(vec![(vec![1], true)]);
+		descriptor.assert_called_with(vec![(vec![1], true), (vec![2], true)]);
 	}
 
 	// Test that descriptor that can't write all bytes returns valid response
@@ -130,27 +135,40 @@ mod tests {
 		let mut queue = OutboundQueue::new(10);
 
 		queue.push_back(vec![1, 2, 3]);
-		assert!(!queue.try_flush_one(&mut descriptor));
+		assert!(!queue.try_flush(&mut descriptor));
 
 		descriptor.assert_called_with(vec![(vec![1, 2, 3], true)]);
 	}
 
+	// Test that descriptor that can't write all bytes (in second pushed item) returns valid response
+	#[test]
+	fn push_back_drain_partial_multiple_push() {
+		let mut descriptor = SocketDescriptorMock::with_fixed_size(2);
+		let mut queue = OutboundQueue::new(10);
+
+		queue.push_back(vec![1]);
+		queue.push_back(vec![2, 3]);
+		assert!(!queue.try_flush(&mut descriptor));
+
+		descriptor.assert_called_with(vec![(vec![1], true), (vec![2, 3], true)]);
+	}
+
 	// Test the bookkeeping for multiple partial writes
 	#[test]
-	fn push_back_drain_partial_drain_partial_try_flush_one() {
+	fn push_back_drain_partial_drain_partial_try_flush() {
 		let mut descriptor = SocketDescriptorMock::with_fixed_size(1);
 		let mut queue = OutboundQueue::new(10);
 
 		queue.push_back(vec![1, 2, 3]);
-		assert!(!queue.try_flush_one(&mut descriptor));
+		assert!(!queue.try_flush(&mut descriptor));
 
 		descriptor.make_room(1);
 		queue.unblock();
-		assert!(!queue.try_flush_one(&mut descriptor));
+		assert!(!queue.try_flush(&mut descriptor));
 
 		descriptor.make_room(1);
 		queue.unblock();
-		assert!(queue.try_flush_one(&mut descriptor));
+		assert!(queue.try_flush(&mut descriptor));
 
 		descriptor.assert_called_with(vec![(vec![1, 2, 3], true), (vec![2, 3], true), (vec![3], true)]);
 	}
@@ -162,27 +180,27 @@ mod tests {
 
 		// Fail write and move to blocked state
 		queue.push_back(vec![1, 2]);
-		assert!(!queue.try_flush_one(&mut descriptor));
+		assert!(!queue.try_flush(&mut descriptor));
 		descriptor.assert_called_with(vec![(vec![1, 2], true)]);
 
 		// Make room but don't signal
 		descriptor.make_room(1);
-		assert!(!queue.try_flush_one(&mut descriptor));
+		assert!(!queue.try_flush(&mut descriptor));
 		assert!(queue.is_blocked());
 		descriptor.assert_called_with(vec![(vec![1, 2], true)]);
 
 		// Unblock and try again
 		queue.unblock();
 
 		// Partial write will succeed, but still move to blocked
-		assert!(!queue.try_flush_one(&mut descriptor));
+		assert!(!queue.try_flush(&mut descriptor));
 		assert!(queue.is_blocked());
 		descriptor.assert_called_with(vec![(vec![1, 2], true), (vec![1, 2], true)]);
 
 		// Make room and signal which will succeed in writing the final piece
 		descriptor.make_room(1);
 		queue.unblock();
-		assert!(queue.try_flush_one(&mut descriptor));
+		assert!(queue.try_flush(&mut descriptor));
 		assert!(!queue.is_blocked());
 		descriptor.assert_called_with(vec![(vec![1, 2], true), (vec![1, 2], true), (vec![2], true)]);
 	}
@@ -194,7 +212,7 @@ mod tests {
 		let mut queue = OutboundQueue::new(1);
 
 		queue.push_back(vec![1]);
-		assert!(queue.try_flush_one(&mut descriptor));
+		assert!(queue.try_flush(&mut descriptor));
 		descriptor.assert_called_with(vec![(vec![1], false)]);
 	}
 
@@ -207,9 +225,7 @@ mod tests {
 		queue.push_back(vec![1]);
 		queue.push_back(vec![2]);
 		queue.push_back(vec![3]);
-		assert!(queue.try_flush_one(&mut descriptor));
-		assert!(queue.try_flush_one(&mut descriptor));
-		assert!(queue.try_flush_one(&mut descriptor));
+		assert!(queue.try_flush(&mut descriptor));
 		descriptor.assert_called_with(vec![(vec![1], false), (vec![2], false), (vec![3], true)]);
 	}
 
@@ -223,7 +239,7 @@ mod tests {
 		queue.push_back(vec![1]);
 		assert!(!queue.is_empty());
 
-		assert!(queue.try_flush_one(&mut descriptor));
+		assert!(queue.try_flush(&mut descriptor));
 		assert!(queue.is_empty());
 	}
 
@@ -244,12 +260,8 @@ mod tests {
 		queue.push_back(vec![2]);
 		assert_eq!(queue.queue_space(), 0);
 
-		// at soft limit
-		assert!(queue.try_flush_one(&mut descriptor));
-		assert_eq!(queue.queue_space(), 0);
-
 		// below soft limt
-		assert!(queue.try_flush_one(&mut descriptor));
+		assert!(queue.try_flush(&mut descriptor));
 		assert_eq!(queue.queue_space(), 1);
 	}
 }