Add a parallel async event handler to OnionMessenger

This adds an `OnionMessenger::process_pending_events_async`
mirroring the same in `ChannelManager`. However, unlike the one in
`ChannelManager`, this processes the events in parallel by spawning
all futures and using the new `MultiFuturePoller`.

Because `OnionMessenger` just generates a stream of messages to
store/fetch, we first process all the events to store new messages,
`await` them, then process all the events to fetch stored messages,
ensuring reordering shouldn't result in lost messages (unless we
race with a peer disconnection, which could happen anyway).

Author: TheBlueMatt

lightning/src/onion_message/messenger.rs

@@ -241,7 +241,7 @@ where
 	offers_handler: OMH,
 	custom_handler: CMH,
 	intercept_messages_for_offline_peers: bool,
-	pending_events: Mutex<Vec<Event>>,
+	pending_events: Mutex<VecDeque<Event>>,
 }
 
 /// [`OnionMessage`]s buffered to be sent.
@@ -963,7 +963,7 @@ where
 			offers_handler,
 			custom_handler,
 			intercept_messages_for_offline_peers,
-			pending_events: Mutex::new(Vec::new()),
+			pending_events: Mutex::new(VecDeque::new()),
 		}
 	}
 
@@ -1146,7 +1146,57 @@ where
 			log_trace!(self.logger, "Dropping event {:?}: buffer full", event);
 			return
 		}
-		pending_events.push(event);
+		pending_events.push_back(event);
+	}
+
+	/// Processes any events asynchronously using the given handler.
+	///
+	/// Note that the event handler is called in the order each event was generated, however
+	/// futures are polled in parallel for some events to allow for parallelism where events do not
+	/// have an ordering requirement.
+	///
+	/// See the trait-level documentation of [`EventsProvider`] for requirements.
+	pub async fn process_pending_events_async<Future: core::future::Future<Output = ()> + core::marker::Unpin, H: Fn(Event) -> Future>(
+		&self, handler: H
+	) {
+		let mut events = VecDeque::new();
+		core::mem::swap(&mut *self.pending_events.lock().unwrap(), &mut events);
+		let mut futures = Vec::with_capacity(events.len());
+		for (node_id, recipient) in self.message_recipients.lock().unwrap().iter_mut() {
+			if let OnionMessageRecipient::PendingConnection(_, addresses, _) = recipient {
+				if let Some(addresses) = addresses.take() {
+					futures.push(Some(handler(Event::ConnectionNeeded { node_id: *node_id, addresses })));
+				}
+			}
+		}
+		// We process events in parallel, but we want to complete `OnionMessageIntercepted` events
+		// prior to `OnionMessagePeerConnected` ones.
+		for _ in 0..events.len() {
+			let next_event = events.pop_front();
+			match next_event {
+				Some(Event::OnionMessageIntercepted { .. }) => {
+					futures.push(Some(handler(next_event.unwrap())));
+				},
+				Some(Event::OnionMessagePeerConnected { .. }) => {
+					events.push_back(next_event.unwrap());
+				},
+				None => debug_assert!(false, "We only iterate events.len() times"),
+				Some(_) => {
+					debug_assert!(false, "Decide on ordering for event type {:?}", next_event);
+					events.push_back(next_event.unwrap());
+				},
+			}
+		}
+		crate::util::async_poll::MultiFuturePoller(futures).await;
+		if events.len() <= 1 {
+			for event in events { handler(event).await; }
+		} else {
+			let mut futures = Vec::new();
+			for event in events {
+				futures.push(Some(handler(event)));
+			}
+			crate::util::async_poll::MultiFuturePoller(futures).await;
+		}
 	}
 }
 
@@ -1192,7 +1242,7 @@ where
 				}
 			}
 		}
-		let mut events = Vec::new();
+		let mut events = VecDeque::new();
 		core::mem::swap(&mut *self.pending_events.lock().unwrap(), &mut events);
 		for ev in events {
 			handler.handle_event(ev);