feat(configurer): send UpdateConfig unboundedly

CHANGELOG.md

@@ -22,6 +22,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 - core/addr: expose the `addr` module.
 - core: generate `NodeNo` randomly if not provided.
 - network: don't force `safe-*` features of the `lz4_flex` crate ([#136]).
+- configurer: send `UpdateConfig` unboundedly.
 
 ### Fixed
 - core: update the `idr-ebr` crate to v0.3 to fix possible crash in `Context::finished()`.

elfo-configurer/src/lib.rs

@@ -16,10 +16,8 @@ use tracing::{error, info, warn};
 
 use elfo_core::{
     config::AnyConfig,
-    errors::RequestError,
     messages::{
-        EntrypointError, Ping, StartEntrypoint, StartEntrypointRejected, UpdateConfig,
-        ValidateConfig,
+        EntrypointError, StartEntrypoint, StartEntrypointRejected, UpdateConfig, ValidateConfig,
     },
     msg, scope,
     signal::{Signal, SignalKind},
@@ -344,27 +342,16 @@ impl Configurer {
     }
 
     async fn update_all(&self, configs: &[ConfigWithMeta]) {
-        let futures = configs
-            .iter()
-            .cloned()
-            .map(|item| {
-                let group = item.group_name;
-                // While `UpdateConfig` is defined as a request to cover more use cases, default
-                // configurer simply sends out new configs instead of waiting for all groups to
-                // process the message and respond. This is done for performance reasons. If an
-                // actor has a congested mailbox or spends a lot of time processing each
-                // message, updating configs using requests can take a lot of time.
-                let fut = self.ctx.send_to(item.addr, UpdateConfig::new(item.config));
-
-                wrap_long_running_future(
-                    fut,
-                    group,
-                    "some actors in the group have congested mailboxes, config update is stalled",
-                )
-            })
-            .collect::<Vec<_>>();
-
-        future::join_all(futures).await;
+        for item in configs {
+            let message = UpdateConfig::new(item.config.clone()); // cheap due to `Arc`s.
+
+            // While `UpdateConfig` is defined as a request to cover more use cases, default
+            // configurer simply sends out new configs instead of waiting for all groups to
+            // process the message and respond. This is done for performance reasons. If an
+            // actor has a congested mailbox or spends a lot of time processing each
+            // message, updating configs using requests can take a lot of time.
+            let _ = self.ctx.unbounded_send_to(item.addr, message);
+        }
     }
 }
 
@@ -384,29 +371,6 @@ async fn wrap_long_running_future<F: Future>(
     }
 }
 
-#[allow(dead_code)]
-async fn ping(ctx: &Context, config_list: &[ConfigWithMeta]) -> bool {
-    let futures = config_list
-        .iter()
-        .cloned()
-        .map(|item| ctx.request_to(item.addr, Ping::default()).all().resolve())
-        .collect::<Vec<_>>();
-
-    // TODO: use `try_join_all`.
-    let errors = future::join_all(futures)
-        .await
-        .into_iter()
-        .flatten()
-        .filter_map(|result| match result {
-            Ok(()) | Err(RequestError::Ignored) => None,
-            Err(RequestError::Failed) => Some(String::from("some group is closed")),
-        })
-        // TODO: include actor keys in the error message.
-        .inspect(|reason| error!(%reason, "ping failed"));
-
-    errors.count() == 0
-}
-
 async fn load_raw_config(path: impl AsRef<Path>) -> Result<Value, String> {
     let content = fs::read_to_string(path)
         .await
@@ -433,6 +397,7 @@ fn match_configs(topology: &Topology, config: &Value) -> Vec<ConfigWithMeta> {
             }
         })
         .collect();
+
     // Config parsing happens in the supervisor, which is executed in this actor
     // when it performs `send_to(addr, UpdateConfig)`. User actor groups can
     // have arbitrary large configs, taking a considerable time to deserialize.