From b8b5f00e9695f46ea30af3ce63aec6dd17f356ae Mon Sep 17 00:00:00 2001 From: hozan23 Date: Thu, 27 Jun 2024 02:39:31 +0200 Subject: Improve async channels error handling and replace unbounded channels with bounded channels Remove all unbounded channels to prevent unbounded memory usage and potential crashes. Use `FuturesUnordered` for sending to multiple channels simultaneously. This prevents the sending loop from blocking if one channel is blocked, and helps handle errors properly. --- core/src/event.rs | 74 +++++++++++++++++++++++++++++++++++++------------------ 1 file changed, 50 insertions(+), 24 deletions(-) (limited to 'core/src/event.rs') diff --git a/core/src/event.rs b/core/src/event.rs index 771d661..1632df3 100644 --- a/core/src/event.rs +++ b/core/src/event.rs @@ -7,17 +7,19 @@ use std::{ use async_channel::{Receiver, Sender}; use chrono::{DateTime, Utc}; +use futures_util::stream::{FuturesUnordered, StreamExt}; use log::{debug, error}; -use crate::{async_runtime::lock::Mutex, util::random_16, Result}; +use crate::{async_runtime::lock::Mutex, util::random_32, Result}; + +const CHANNEL_BUFFER_SIZE: usize = 1000; pub type ArcEventSys = Arc>; -pub type WeakEventSys = Weak>; -pub type EventListenerID = u16; +pub type EventListenerID = u32; type Listeners = HashMap>>>; -/// EventSys supports event emission to registered listeners based on topics. +/// EventSys emits events to registered listeners based on topics. /// # Example /// /// ``` @@ -74,22 +76,41 @@ type Listeners = HashMap { listeners: Mutex>, + listener_buffer_size: usize, } impl EventSys where T: std::hash::Hash + Eq + std::fmt::Debug + Clone, { - /// Creates a new `EventSys` + /// Creates a new [`EventSys`] pub fn new() -> ArcEventSys { Arc::new(Self { listeners: Mutex::new(HashMap::new()), + listener_buffer_size: CHANNEL_BUFFER_SIZE, + }) + } + + /// Creates a new [`EventSys`] with the provided buffer size for the + /// [`EventListener`] channel. + /// + /// This is important to control the memory used by the listener channel. + /// If the consumer for the event listener can't keep up with the new events + /// coming, then the channel buffer will fill with new events, and if the + /// buffer is full, the emit function will block until the listener + /// starts to consume the buffered events. + /// + /// If `size` is zero, this function will panic. + pub fn with_buffer_size(size: usize) -> ArcEventSys { + Arc::new(Self { + listeners: Mutex::new(HashMap::new()), + listener_buffer_size: size, }) } /// Emits an event to the listeners. /// - /// The event must implement the `EventValueTopic` trait to indicate the + /// The event must implement the [`EventValueTopic`] trait to indicate the /// topic of the event. Otherwise, you can use `emit_by_topic()`. pub async fn emit + Clone>(&self, value: &E) { let topic = E::topic(); @@ -115,22 +136,26 @@ where let event_id = E::id().to_string(); if !event_ids.contains_key(&event_id) { - debug!( - "Failed to emit an event to a non-existent event id: {:?}", - event_id - ); + debug!("Failed to emit an event: unknown event id {:?}", event_id); return; } - let mut failed_listeners = vec![]; + let mut results = FuturesUnordered::new(); let listeners = event_ids.get_mut(&event_id).unwrap(); for (listener_id, listener) in listeners.iter() { - if let Err(err) = listener.send(event.clone()).await { + let result = async { (*listener_id, listener.send(event.clone()).await) }; + results.push(result); + } + + let mut failed_listeners = vec![]; + while let Some((id, fut_err)) = results.next().await { + if let Err(err) = fut_err { debug!("Failed to emit event for topic {:?}: {}", topic, err); - failed_listeners.push(*listener_id); + failed_listeners.push(id); } } + drop(results); for listener_id in failed_listeners.iter() { listeners.remove(listener_id); @@ -142,7 +167,7 @@ where self: &Arc, topic: &T, ) -> EventListener { - let chan = async_channel::unbounded(); + let chan = async_channel::bounded(self.listener_buffer_size); let topics = &mut self.listeners.lock().await; @@ -159,9 +184,10 @@ where let listeners = event_ids.get_mut(&event_id).unwrap(); - let mut listener_id = random_16(); + let mut listener_id = random_32(); + // Generate a new one if listener_id already exists while listeners.contains_key(&listener_id) { - listener_id = random_16(); + listener_id = random_32(); } let listener = @@ -197,7 +223,7 @@ where pub struct EventListener { id: EventListenerID, recv_chan: Receiver, - event_sys: WeakEventSys, + event_sys: Weak>, event_id: String, topic: T, phantom: PhantomData, @@ -208,10 +234,10 @@ where T: std::hash::Hash + Eq + Clone + std::fmt::Debug, E: EventValueAny + Clone + EventValue, { - /// Create a new event listener. + /// Creates a new [`EventListener`]. fn new( id: EventListenerID, - event_sys: WeakEventSys, + event_sys: Weak>, recv_chan: Receiver, event_id: &str, topic: &T, @@ -226,12 +252,12 @@ where } } - /// Receive the next event. + /// Receives the next event. pub async fn recv(&self) -> Result { match self.recv_chan.recv().await { Ok(event) => match ((*event.value).value_as_any()).downcast_ref::() { Some(v) => Ok(v.clone()), - None => unreachable!("Error when attempting to downcast the event value."), + None => unreachable!("Failed to downcast the event value."), }, Err(err) => { error!("Failed to receive new event: {err}"); @@ -241,7 +267,7 @@ where } } - /// Cancels the listener and removes it from the `EventSys`. + /// Cancels the event listener and removes it from the [`EventSys`]. pub async fn cancel(&self) { if let Some(es) = self.event_sys.upgrade() { es.remove(&self.topic, &self.event_id, &self.id).await; @@ -249,12 +275,12 @@ where } /// Returns the topic for this event listener. - pub async fn topic(&self) -> &T { + pub fn topic(&self) -> &T { &self.topic } /// Returns the event id for this event listener. - pub async fn event_id(&self) -> &String { + pub fn event_id(&self) -> &String { &self.event_id } } -- cgit v1.2.3