Introduce EventQueueNotifierGuard type

tnull · tnull · commit 588d2fa25896 · 2025-02-12T17:06:40.000+01: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.rs b/lightning-liquidity/src/events.rs
@@ -31,7 +31,7 @@ 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>,
 }
 
 impl EventQueue {
@@ -40,7 +40,7 @@ impl EventQueue {
 		let waker = Arc::new(Mutex::new(None));
 		#[cfg(feature = "std")]
 		{
-			let condvar = crate::sync::Condvar::new();
+			let condvar = Arc::new(crate::sync::Condvar::new());
 			Self { queue, waker, condvar }
 		}
 		#[cfg(not(feature = "std"))]
@@ -100,6 +100,49 @@ 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(feature = "std")]
+		{
+			EventQueueNotifierGuard {
+				queue: Arc::clone(&self.queue),
+				waker: Arc::clone(&self.waker),
+				condvar: Arc::clone(&self.condvar),
+			}
+		}
+		#[cfg(not(feature = "std"))]
+		{
+			EventQueueNotifierGuard {
+				queue: Arc::clone(&self.queue),
+				waker: Arc::clone(&self.waker),
+			}
+		}
+	}
+}
+
+// A guard type that will notify about new events when dropped.
+#[must_use]
+pub(crate) struct EventQueueNotifierGuard {
+	queue: Arc<Mutex<VecDeque<Event>>>,
+	waker: Arc<Mutex<Option<Waker>>>,
+	#[cfg(feature = "std")]
+	condvar: Arc<crate::sync::Condvar>,
+}
+
+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();
+		}
+	}
 }
 
 /// An event which you should probably take some action in response to.
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
@@ -105,8 +105,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();
@@ -142,6 +143,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) => {
@@ -219,6 +222,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) => {
@@ -261,6 +266,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) => {
@@ -338,6 +345,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) => {
@@ -380,6 +389,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
@@ -184,6 +184,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) => {
@@ -250,6 +252,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
@@ -782,6 +782,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();
@@ -1071,6 +1073,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 =
@@ -1090,7 +1093,6 @@ where
 						token: params.token,
 					};
 					self.pending_events.enqueue(event);
-
 					(Ok(()), msg)
 				},
 				(e, msg) => (e, msg),
@@ -1107,6 +1109,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 {
