Introduce EventQueueNotifierGuard type

tnull · tnull · commit 8cfa0aeadc6f · 2025-04-15T10:47:28.000+02:00
Previously, when enqueuing new events to the `EventQueue`, we'd
directly attempt to wake any notifiers/notify any threads waiting on
the `Condvar` about the newly available events. This could of course
mean we'd notify them while ourselves still holding some locks, e.g., on
the peer state.

Here, we instead introduce a `EventQueueNotifierGuard` type that will
notify about pending events if necesssary, which mitigates any potential
lock contention: we now simply have to ensure that any method calling
`enqueue` holds the notifier before retrieving any locks.
diff --git a/lightning-liquidity/src/events/event_queue.rs b/lightning-liquidity/src/events/event_queue.rs
@@ -5,6 +5,8 @@ use alloc::collections::VecDeque;
 use alloc::vec::Vec;
 
 use core::future::Future;
+#[cfg(debug_assertions)]
+use core::sync::atomic::{AtomicU8, Ordering};
 use core::task::{Poll, Waker};
 
 /// The maximum queue size we allow before starting to drop events.
@@ -14,37 +16,40 @@ pub(crate) struct EventQueue {
 	queue: Arc<Mutex<VecDeque<LiquidityEvent>>>,
 	waker: Arc<Mutex<Option<Waker>>>,
 	#[cfg(feature = "std")]
-	condvar: crate::sync::Condvar,
+	condvar: Arc<crate::sync::Condvar>,
+	#[cfg(debug_assertions)]
+	num_held_notifier_guards: Arc<AtomicU8>,
 }
 
 impl EventQueue {
 	pub fn new() -> Self {
 		let queue = Arc::new(Mutex::new(VecDeque::new()));
 		let waker = Arc::new(Mutex::new(None));
-		#[cfg(feature = "std")]
-		{
-			let condvar = crate::sync::Condvar::new();
-			Self { queue, waker, condvar }
+		Self {
+			queue,
+			waker,
+			#[cfg(feature = "std")]
+			condvar: Arc::new(crate::sync::Condvar::new()),
+			#[cfg(debug_assertions)]
+			num_held_notifier_guards: Arc::new(AtomicU8::new(0)),
 		}
-		#[cfg(not(feature = "std"))]
-		Self { queue, waker }
 	}
 
 	pub fn enqueue<E: Into<LiquidityEvent>>(&self, event: E) {
+		#[cfg(debug_assertions)]
 		{
-			let mut queue = self.queue.lock().unwrap();
-			if queue.len() < MAX_EVENT_QUEUE_SIZE {
-				queue.push_back(event.into());
-			} else {
-				return;
-			}
+			let num_held_notifier_guards = self.num_held_notifier_guards.load(Ordering::Relaxed);
+			debug_assert!(
+				num_held_notifier_guards > 0,
+				"We should be holding at least one notifier guard whenever enqueuing new events"
+			);
 		}
-
-		if let Some(waker) = self.waker.lock().unwrap().take() {
-			waker.wake();
+		let mut queue = self.queue.lock().unwrap();
+		if queue.len() < MAX_EVENT_QUEUE_SIZE {
+			queue.push_back(event.into());
+		} else {
+			return;
 		}
-		#[cfg(feature = "std")]
-		self.condvar.notify_one();
 	}
 
 	pub fn next_event(&self) -> Option<LiquidityEvent> {
@@ -83,6 +88,81 @@ impl EventQueue {
 	pub fn get_and_clear_pending_events(&self) -> Vec<LiquidityEvent> {
 		self.queue.lock().unwrap().split_off(0).into()
 	}
+
+	// Returns an [`EventQueueNotifierGuard`] that will notify about new event when dropped.
+	pub fn notifier(&self) -> EventQueueNotifierGuard {
+		#[cfg(debug_assertions)]
+		{
+			self.num_held_notifier_guards.fetch_add(1, Ordering::Relaxed);
+		}
+		EventQueueNotifierGuard {
+			queue: Arc::clone(&self.queue),
+			waker: Arc::clone(&self.waker),
+			#[cfg(feature = "std")]
+			condvar: Arc::clone(&self.condvar),
+			#[cfg(debug_assertions)]
+			num_held_notifier_guards: Arc::clone(&self.num_held_notifier_guards),
+		}
+	}
+}
+
+impl Drop for EventQueue {
+	fn drop(&mut self) {
+		#[cfg(debug_assertions)]
+		{
+			let num_held_notifier_guards = self.num_held_notifier_guards.load(Ordering::Relaxed);
+			debug_assert!(
+				num_held_notifier_guards == 0,
+				"We should not be holding any notifier guards when the event queue is dropped"
+			);
+		}
+	}
+}
+
+// A guard type that will notify about new events when dropped.
+#[must_use]
+pub(crate) struct EventQueueNotifierGuard {
+	queue: Arc<Mutex<VecDeque<LiquidityEvent>>>,
+	waker: Arc<Mutex<Option<Waker>>>,
+	#[cfg(feature = "std")]
+	condvar: Arc<crate::sync::Condvar>,
+	#[cfg(debug_assertions)]
+	num_held_notifier_guards: Arc<AtomicU8>,
+}
+
+impl Drop for EventQueueNotifierGuard {
+	fn drop(&mut self) {
+		let should_notify = !self.queue.lock().unwrap().is_empty();
+
+		if should_notify {
+			if let Some(waker) = self.waker.lock().unwrap().take() {
+				waker.wake();
+			}
+
+			#[cfg(feature = "std")]
+			self.condvar.notify_one();
+		}
+
+		#[cfg(debug_assertions)]
+		{
+			let res = self.num_held_notifier_guards.fetch_update(
+				Ordering::Relaxed,
+				Ordering::Relaxed,
+				|x| Some(x.saturating_sub(1)),
+			);
+			match res {
+				Ok(previous_value) if previous_value == 0 => debug_assert!(
+					false,
+					"num_held_notifier_guards counter out-of-sync! This should never happen!"
+				),
+				Err(_) => debug_assert!(
+					false,
+					"num_held_notifier_guards counter out-of-sync! This should never happen!"
+				),
+				_ => {},
+			}
+		}
+	}
 }
 
 struct EventFuture {
@@ -129,6 +209,7 @@ mod tests {
 		});
 
 		for _ in 0..3 {
+			let _guard = event_queue.notifier();
 			event_queue.enqueue(expected_event.clone());
 		}
 
@@ -154,13 +235,15 @@ mod tests {
 		let mut delayed_enqueue = false;
 
 		for _ in 0..25 {
+			let _guard = event_queue.notifier();
 			event_queue.enqueue(expected_event.clone());
 			enqueued_events.fetch_add(1, Ordering::SeqCst);
 		}
 
 		loop {
 			tokio::select! {
 				_ = tokio::time::sleep(Duration::from_millis(10)), if !delayed_enqueue => {
+					let _guard = event_queue.notifier();
 					event_queue.enqueue(expected_event.clone());
 					enqueued_events.fetch_add(1, Ordering::SeqCst);
 					delayed_enqueue = true;
@@ -169,6 +252,7 @@ mod tests {
 					assert_eq!(e, expected_event);
 					received_events.fetch_add(1, Ordering::SeqCst);
 
+					let _guard = event_queue.notifier();
 					event_queue.enqueue(expected_event.clone());
 					enqueued_events.fetch_add(1, Ordering::SeqCst);
 				}
@@ -201,6 +285,7 @@ mod tests {
 		std::thread::spawn(move || {
 			// Sleep a bit before we enqueue the events everybody is waiting for.
 			std::thread::sleep(Duration::from_millis(20));
+			let _guard = thread_queue.notifier();
 			thread_queue.enqueue(thread_event.clone());
 			thread_queue.enqueue(thread_event.clone());
 		});
diff --git a/lightning-liquidity/src/lsps0/client.rs b/lightning-liquidity/src/lsps0/client.rs
@@ -61,6 +61,8 @@ where
 	fn handle_response(
 		&self, response: LSPS0Response, counterparty_node_id: &PublicKey,
 	) -> Result<(), LightningError> {
+		let _event_queue_notifier = self.pending_events.notifier();
+
 		match response {
 			LSPS0Response::ListProtocols(LSPS0ListProtocolsResponse { protocols }) => {
 				self.pending_events.enqueue(LSPS0ClientEvent::ListProtocolsResponse {
diff --git a/lightning-liquidity/src/lsps1/client.rs b/lightning-liquidity/src/lsps1/client.rs
@@ -110,8 +110,9 @@ where
 		&self, request_id: LSPSRequestId, counterparty_node_id: &PublicKey,
 		result: LSPS1GetInfoResponse,
 	) -> Result<(), LightningError> {
-		let outer_state_lock = self.per_peer_state.write().unwrap();
+		let _event_queue_notifier = self.pending_events.notifier();
 
+		let outer_state_lock = self.per_peer_state.write().unwrap();
 		match outer_state_lock.get(counterparty_node_id) {
 			Some(inner_state_lock) => {
 				let mut peer_state_lock = inner_state_lock.lock().unwrap();
@@ -147,6 +148,8 @@ where
 		&self, request_id: LSPSRequestId, counterparty_node_id: &PublicKey,
 		error: LSPSResponseError,
 	) -> Result<(), LightningError> {
+		let _event_queue_notifier = self.pending_events.notifier();
+
 		let outer_state_lock = self.per_peer_state.read().unwrap();
 		match outer_state_lock.get(counterparty_node_id) {
 			Some(inner_state_lock) => {
@@ -224,6 +227,8 @@ where
 		&self, request_id: LSPSRequestId, counterparty_node_id: &PublicKey,
 		response: LSPS1CreateOrderResponse,
 	) -> Result<(), LightningError> {
+		let _event_queue_notifier = self.pending_events.notifier();
+
 		let outer_state_lock = self.per_peer_state.read().unwrap();
 		match outer_state_lock.get(counterparty_node_id) {
 			Some(inner_state_lock) => {
@@ -266,6 +271,8 @@ where
 		&self, request_id: LSPSRequestId, counterparty_node_id: &PublicKey,
 		error: LSPSResponseError,
 	) -> Result<(), LightningError> {
+		let _event_queue_notifier = self.pending_events.notifier();
+
 		let outer_state_lock = self.per_peer_state.read().unwrap();
 		match outer_state_lock.get(counterparty_node_id) {
 			Some(inner_state_lock) => {
@@ -343,6 +350,8 @@ where
 		&self, request_id: LSPSRequestId, counterparty_node_id: &PublicKey,
 		response: LSPS1CreateOrderResponse,
 	) -> Result<(), LightningError> {
+		let _event_queue_notifier = self.pending_events.notifier();
+
 		let outer_state_lock = self.per_peer_state.read().unwrap();
 		match outer_state_lock.get(counterparty_node_id) {
 			Some(inner_state_lock) => {
@@ -385,6 +394,8 @@ where
 		&self, request_id: LSPSRequestId, counterparty_node_id: &PublicKey,
 		error: LSPSResponseError,
 	) -> Result<(), LightningError> {
+		let _event_queue_notifier = self.pending_events.notifier();
+
 		let outer_state_lock = self.per_peer_state.read().unwrap();
 		match outer_state_lock.get(counterparty_node_id) {
 			Some(inner_state_lock) => {
diff --git a/lightning-liquidity/src/lsps2/client.rs b/lightning-liquidity/src/lsps2/client.rs
@@ -191,6 +191,8 @@ where
 		&self, request_id: LSPSRequestId, counterparty_node_id: &PublicKey,
 		result: LSPS2GetInfoResponse,
 	) -> Result<(), LightningError> {
+		let _event_queue_notifier = self.pending_events.notifier();
+
 		let outer_state_lock = self.per_peer_state.read().unwrap();
 		match outer_state_lock.get(counterparty_node_id) {
 			Some(inner_state_lock) => {
@@ -257,6 +259,8 @@ where
 		&self, request_id: LSPSRequestId, counterparty_node_id: &PublicKey,
 		result: LSPS2BuyResponse,
 	) -> Result<(), LightningError> {
+		let _event_queue_notifier = self.pending_events.notifier();
+
 		let outer_state_lock = self.per_peer_state.read().unwrap();
 		match outer_state_lock.get(counterparty_node_id) {
 			Some(inner_state_lock) => {
diff --git a/lightning-liquidity/src/lsps2/service.rs b/lightning-liquidity/src/lsps2/service.rs
@@ -777,6 +777,8 @@ where
 		&self, intercept_scid: u64, intercept_id: InterceptId, expected_outbound_amount_msat: u64,
 		payment_hash: PaymentHash,
 	) -> Result<(), APIError> {
+		let _event_queue_notifier = self.pending_events.notifier();
+
 		let peer_by_intercept_scid = self.peer_by_intercept_scid.read().unwrap();
 		if let Some(counterparty_node_id) = peer_by_intercept_scid.get(&intercept_scid) {
 			let outer_state_lock = self.per_peer_state.read().unwrap();
@@ -1066,6 +1068,7 @@ where
 		&self, request_id: LSPSRequestId, counterparty_node_id: &PublicKey,
 		params: LSPS2GetInfoRequest,
 	) -> Result<(), LightningError> {
+		let _event_queue_notifier = self.pending_events.notifier();
 		let (result, response) = {
 			let mut outer_state_lock = self.per_peer_state.write().unwrap();
 			let inner_state_lock =
@@ -1085,7 +1088,6 @@ where
 						token: params.token,
 					};
 					self.pending_events.enqueue(event);
-
 					(Ok(()), msg)
 				},
 				(e, msg) => (e, msg),
@@ -1102,6 +1104,7 @@ where
 	fn handle_buy_request(
 		&self, request_id: LSPSRequestId, counterparty_node_id: &PublicKey, params: LSPS2BuyRequest,
 	) -> Result<(), LightningError> {
+		let _event_queue_notifier = self.pending_events.notifier();
 		if let Some(payment_size_msat) = params.payment_size_msat {
 			if payment_size_msat < params.opening_fee_params.min_payment_size_msat {
 				let response = LSPS2Response::BuyError(LSPSResponseError {
