remove futures buffered

lherman-cs · lherman-cs · commit caa0f9314c5c · 2025-12-03T12:31:20.000-05:00
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/pulsebeam/Cargo.toml b/pulsebeam/Cargo.toml
@@ -39,7 +39,6 @@ metrics-exporter-prometheus = "0.17.2"
 pprof = { version = "0.15.0", features = ["flamegraph", "prost-codec", "protobuf"] }
 tokio-metrics = { version = "0.4.5", features = ["metrics-rs-integration"] }
 once_cell = "1.21.3"
-futures-buffered = "0.2.12"
 ahash = "0.8.12"
 
 [dev-dependencies]
diff --git a/pulsebeam/src/shard.rs b/pulsebeam/src/shard.rs
@@ -1,7 +1,6 @@
 use std::pin::Pin;
 
-use futures::StreamExt;
-use futures_buffered::FuturesUnorderedBounded;
+use futures::{StreamExt, stream::FuturesUnordered};
 use pulsebeam_runtime::{actor, sync::Arc};
 
 pub type ShardTask = Pin<Box<dyn futures::Future<Output = ()> + Send>>;
@@ -12,7 +11,7 @@ pub enum ShardMessage {
 
 pub struct ShardActor {
     shard_id: usize,
-    tasks: FuturesUnorderedBounded<ShardTask>,
+    tasks: FuturesUnordered<ShardTask>,
 }
 
 pub struct ShardMessageSet;
@@ -70,7 +69,7 @@ impl ShardActor {
     pub fn new(shard_id: usize) -> Self {
         Self {
             shard_id,
-            tasks: FuturesUnorderedBounded::new(512),
+            tasks: FuturesUnordered::new(),
         }
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -1,7 +1,6 @@`
`1`	`1`	`use std::pin::Pin;`
`2`	`2`
`3`		`-use futures::StreamExt;`
`4`		`-use futures_buffered::FuturesUnorderedBounded;`
	`3`	`+use futures::{StreamExt, stream::FuturesUnordered};`
`5`	`4`	`use pulsebeam_runtime::{actor, sync::Arc};`
`6`	`5`
`7`	`6`	`pub type ShardTask = Pin<Box<dyn futures::Future<Output = ()> + Send>>;`
`@@ -12,7 +11,7 @@ pub enum ShardMessage {`
`12`	`11`
`13`	`12`	`pub struct ShardActor {`
`14`	`13`	`shard_id: usize,`
`15`		`- tasks: FuturesUnorderedBounded<ShardTask>,`
	`14`	`+ tasks: FuturesUnordered<ShardTask>,`
`16`	`15`	`}`
`17`	`16`
`18`	`17`	`pub struct ShardMessageSet;`
`@@ -70,7 +69,7 @@ impl ShardActor {`
`70`	`69`	`pub fn new(shard_id: usize) -> Self {`
`71`	`70`	`Self {`
`72`	`71`	`shard_id,`
`73`		`- tasks: FuturesUnorderedBounded::new(512),`
	`72`	`+ tasks: FuturesUnordered::new(),`
`74`	`73`	`}`
`75`	`74`	`}`
`76`	`75`	`}`