Skip to content

perf(pm): stream install_packages layer drain via FuturesUnordered #2904

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
elrrrrrrr wants to merge 8 commits into from
Closed

perf(pm): stream install_packages layer drain via FuturesUnordered #2904

Changes from all commits
Commits
Show all changes
8 commits
Select commit Hold shift + click to select a range
e1f3f18
perf(pm): stream install_packages layer drain via FuturesUnordered
elrrrrrrr May 7, 2026
428e8a0
Merge remote-tracking branch 'origin/next' into experiment/install-st…
elrrrrrrr May 7, 2026
a7483d4
chore(bench): extend pcap to install phase + utoo-next baseline
elrrrrrrr May 7, 2026
d5e0d62
chore(bench): add tshark TCP-stress metrics to pcap bench
elrrrrrrr May 7, 2026
5abfd41
fix(bench): make tshark analysis best-effort, awk-only wall-time parse
elrrrrrrr May 7, 2026
06d8203
chore(bench): add iostat sampling to pcap bench for disk-saturation d…
elrrrrrrr May 7, 2026
4541d7f
Merge remote-tracking branch 'origin/next' into experiment/install-st…
elrrrrrrr May 8, 2026
2d54cc6
Merge branch 'next' into experiment/install-streaming-drain
elrrrrrrr May 8, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 11 additions & 3 deletions crates/pm/src/service/install.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
use crate::util::cli_enum::ScriptPolicy;
use anyhow::Context;
use anyhow::Result;
use futures::stream::{FuturesUnordered, StreamExt};
use std::collections::{HashMap, HashSet};
use std::path::Path;
use std::time::Instant;
Expand Down Expand Up @@ -83,7 +84,14 @@ pub async fn install_packages(
depths.sort_unstable();

for depth in depths.iter() {
let mut clone_tasks: Vec<tokio::task::JoinHandle<Result<()>>> = Vec::new();
// Stream task completion within a layer instead of joining serially.
// The previous `for task in clone_tasks { task.await?? }` walks tasks
// in spawn order — slow tasks block the pop of already-finished
// ones, leaving idle latency between layers on deep trees. With
// FuturesUnordered we drain in completion order, propagating the
// first error early.
let mut clone_tasks: FuturesUnordered<tokio::task::JoinHandle<Result<()>>> =
FuturesUnordered::new();
Comment on lines +93 to +94

@gemini-code-assist gemini-code-assist Bot May 7, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Consider using tokio::task::JoinSet instead of FuturesUnordered<tokio::task::JoinHandle<_>>.

While FuturesUnordered allows for processing completions in any order, it has a significant drawback in this context: when the FuturesUnordered collection is dropped (e.g., when an error is propagated via ? in the drain loop), the contained JoinHandles are dropped, which detaches the tasks rather than cancelling them. This means background installation tasks will continue to run even after an error has been returned to the caller.

tokio::task::JoinSet is more efficient for this pattern and, crucially, aborts all remaining tasks when dropped. This ensures that if one package fails, all other pending clones in the same layer are immediately cancelled, saving resources and avoiding orphaned background work.

Note: Implementing this would also involve replacing tokio::spawn(...) and clone_tasks.push(task) with clone_tasks.spawn(...).

References
  1. Resource management and task lifecycle: prefer JoinSet over FuturesUnordered of JoinHandles when cancellation on error is desired, as JoinHandles detach on drop.


if let Some(packages) = groups.get(depth) {
for (path, package) in packages.iter() {
Expand Down Expand Up @@ -169,8 +177,8 @@ pub async fn install_packages(
}
}

for task in clone_tasks {
task.await??;
while let Some(joined) = clone_tasks.next().await {
joined??;
}
}

Expand Down
Loading