Introduce EventQueueNotifierGuard type

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.

Author: tnull

lightning-liquidity/src/events.rs

@@ -31,7 +31,7 @@ pub(crate) struct EventQueue {
 	queue: Arc<Mutex<VecDeque<Event>>>,
 	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"))]
@@ -98,6 +98,49 @@ impl EventQueue {
 	pub fn get_and_clear_pending_events(&self) -> Vec<Event> {
 		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.

lightning-liquidity/src/lsps0/client.rs

@@ -60,6 +60,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(ListProtocolsResponse { protocols }) => {
 				self.pending_events.enqueue(Event::LSPS0Client(

lightning-liquidity/src/lsps1/client.rs

@@ -103,8 +103,9 @@ where
 	fn handle_get_info_response(
 		&self, request_id: RequestId, counterparty_node_id: &PublicKey, result: GetInfoResponse,
 	) -> 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();
@@ -141,6 +142,8 @@ where
 	fn handle_get_info_error(
 		&self, request_id: RequestId, counterparty_node_id: &PublicKey, error: ResponseError,
 	) -> 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) => {
@@ -218,6 +221,8 @@ where
 		&self, request_id: RequestId, counterparty_node_id: &PublicKey,
 		response: CreateOrderResponse,
 	) -> 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) => {
@@ -259,6 +264,8 @@ where
 	fn handle_create_order_error(
 		&self, request_id: RequestId, counterparty_node_id: &PublicKey, error: ResponseError,
 	) -> 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) => {
@@ -337,6 +344,8 @@ where
 		&self, request_id: RequestId, counterparty_node_id: &PublicKey,
 		response: CreateOrderResponse,
 	) -> 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) => {
@@ -378,6 +387,8 @@ where
 	fn handle_get_order_error(
 		&self, request_id: RequestId, counterparty_node_id: &PublicKey, error: ResponseError,
 	) -> 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) => {

lightning-liquidity/src/lsps2/client.rs

@@ -183,6 +183,8 @@ where
 	fn handle_get_info_response(
 		&self, request_id: RequestId, counterparty_node_id: &PublicKey, result: GetInfoResponse,
 	) -> 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) => {
@@ -249,6 +251,8 @@ where
 	fn handle_buy_response(
 		&self, request_id: RequestId, counterparty_node_id: &PublicKey, result: BuyResponse,
 	) -> 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) => {

lightning-liquidity/src/lsps2/service.rs

@@ -790,6 +790,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();
@@ -1078,6 +1080,7 @@ where
 	fn handle_get_info_request(
 		&self, request_id: RequestId, counterparty_node_id: &PublicKey, params: GetInfoRequest,
 	) -> 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 =
@@ -1097,7 +1100,6 @@ where
 						token: params.token,
 					});
 					self.pending_events.enqueue(event);
-
 					(Ok(()), msg)
 				},
 				(e, msg) => (e, msg),
@@ -1114,6 +1116,7 @@ where
 	fn handle_buy_request(
 		&self, request_id: RequestId, counterparty_node_id: &PublicKey, params: BuyRequest,
 	) -> 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(ResponseError {