use join set instead of futuresunordered

Author: lherman-cs

pulsebeam/src/main.rs

@@ -29,6 +29,7 @@ fn main() {
     let rt = tokio::runtime::Builder::new_multi_thread()
         .enable_all()
         .worker_threads(workers)
+        .disable_lifo_slot()
         .build()
         .unwrap();
     // let rt = tokio::runtime::LocalRuntime::new().unwrap();

pulsebeam/src/node.rs

@@ -77,7 +77,7 @@ pub async fn run(
     let (gateway, gateway_join) = actor::spawn_default(gateway::GatewayActor::new(sockets.clone()));
     join_set.push(gateway_join.map(|_| ()).boxed());
 
-    let shard_count = 16 * workers;
+    let shard_count = 2 * workers;
     let mut shard_handles = Vec::with_capacity(shard_count);
 
     for i in 0..shard_count {

pulsebeam/src/shard.rs

@@ -1,7 +1,7 @@
 use std::pin::Pin;
 
-use futures::{StreamExt, stream::FuturesUnordered};
 use pulsebeam_runtime::{actor, sync::Arc};
+use tokio::task::JoinSet;
 
 pub type ShardTask = Pin<Box<dyn futures::Future<Output = ()> + Send>>;
 
@@ -11,7 +11,7 @@ pub enum ShardMessage {
 
 pub struct ShardActor {
     shard_id: usize,
-    tasks: FuturesUnordered<ShardTask>,
+    tasks: JoinSet<()>,
 }
 
 pub struct ShardMessageSet;
@@ -48,7 +48,7 @@ impl actor::Actor<ShardMessageSet> for ShardActor {
         pulsebeam_runtime::actor_loop!(self, _ctx,
             pre_select: {},
             select: {
-                Some(_) = self.tasks.next() => {
+                Some(_) = self.tasks.join_next() => {
                     // task completed, shard does nothing else
                 }
             }
@@ -59,7 +59,7 @@ impl actor::Actor<ShardMessageSet> for ShardActor {
 
     async fn on_msg(&mut self, _ctx: &mut actor::ActorContext<ShardMessageSet>, msg: ShardMessage) {
         let ShardMessage::AddTask(task) = msg;
-        self.tasks.push(task);
+        self.tasks.spawn(task);
         metrics::counter!("shard_task_count", "shard_id" => self.shard_id.to_string())
             .absolute(self.tasks.len() as u64);
     }
@@ -69,7 +69,7 @@ impl ShardActor {
     pub fn new(shard_id: usize) -> Self {
         Self {
             shard_id,
-            tasks: FuturesUnordered::new(),
+            tasks: JoinSet::new(),
         }
     }
 }