perf(acton-reactive): optimize broadcast with pre-allocated Vec and increased inbox capacity

- Replace FuturesUnordered with Vec::with_capacity() + join_all() for
  broadcast to enable pre-allocation and reduce memory churn
- Increase default actor_inbox_capacity from 255 to 512 for improved
  throughput in high-volume messaging scenarios

Author: rrrodzilla

acton-reactive/src/common/broker.rs

@@ -21,7 +21,7 @@ use std::sync::Arc;
 
 use acton_ern::Ern;
 use dashmap::DashMap;
-use futures::stream::{FuturesUnordered, StreamExt};
+use futures::future::join_all;
 use tracing::{instrument, trace};
 
 use crate::actor::{ActorConfig, Idle, ManagedActor};
@@ -205,25 +205,22 @@ impl Broker {
             // BrokerRequestEnvelope contains Arc<dyn ActonMessage>, so cloning is cheap.
             let shared_envelope: BrokerRequestEnvelope = request.into();
 
-            // Use FuturesUnordered to execute sends in parallel rather than sequentially.
-            // This allows the Tokio runtime to poll all futures concurrently.
-            let mut futures: FuturesUnordered<_> = subscribers_set
-                .value()
-                .iter()
-                .map(|(_, subscriber_handle)| {
-                    let handle = subscriber_handle.clone();
-                    let envelope_to_send = shared_envelope.clone();
-                    async move {
-                        trace!(subscriber = %handle.id(), message_type = ?message_type_id, "Sending broadcast");
-                        // Send the envelope to the subscriber's handle.
-                        // Ignore potential send errors (e.g., closed channel).
-                        handle.send(envelope_to_send).await;
-                    }
-                })
-                .collect();
+            // Pre-allocate Vec with known capacity, then use join_all for parallel execution.
+            // This avoids reallocations during iteration and executes all sends concurrently.
+            let mut futures = Vec::with_capacity(num_subscribers);
+            for (_, subscriber_handle) in subscribers_set.value() {
+                let handle = subscriber_handle.clone();
+                let envelope_to_send = shared_envelope.clone();
+                futures.push(async move {
+                    trace!(subscriber = %handle.id(), message_type = ?message_type_id, "Sending broadcast");
+                    // Send the envelope to the subscriber's handle.
+                    // Ignore potential send errors (e.g., closed channel).
+                    handle.send(envelope_to_send).await;
+                });
+            }
 
-            // Drive all futures to completion in parallel.
-            while futures.next().await.is_some() {}
+            // Execute all sends concurrently and wait for completion.
+            join_all(futures).await;
 
             trace!(count = num_subscribers, message_type = ?message_type_id, "Broadcast sends completed");
         } else {

acton-reactive/src/common/config.rs

@@ -118,7 +118,7 @@ impl Default for LimitsConfig {
     fn default() -> Self {
         Self {
             concurrent_handlers_high_water_mark: 100,
-            actor_inbox_capacity: 255,
+            actor_inbox_capacity: 512,
             dummy_channel_size: 1,
         }
     }