Rust structured concurrency: cap media decoders w/ tokio JoinSet

Why structured concurrency for media decode

Media decode pipelines (transcoding with ffmpeg, waveform and thumbnail extraction) push you toward fan-out. Without guardrails you get unbounded task creation, stray subprocesses, and shutdowns that depend on timing. Rust with tokio has the right building blocks:

• JoinSet spawns tasks tied to a scope; dropping it aborts any unfinished tasks.
• Semaphore sets a hard concurrency cap and applies backpressure.
• CancellationToken with kill_on_drop and start_kill enables cooperative, fast shutdown of child processes.

Below is a practical pattern for capping parallel decoders with JoinSet plus a tokio Semaphore, and for making ffmpeg wrappers cancellation-safe.

JoinSet vs FuturesUnordered (and friends)

• JoinSet creates real tasks, and dropping the JoinSet aborts those still running. That fits “these workers belong to this scope.”
• FuturesUnordered drives futures you supply. If you call tokio::spawn inside it, you’ve detached work and must manage aborts and joins yourself. If you don’t spawn, the futures must be !Send-aware and cannot perform blocking or syscall-heavy work inline.

For decode workers with structure, default to JoinSet.

Pattern 1: Bounded fan-out with owned semaphore permits

Two rules keep the cap honest:

1. Acquire a permit before spawning. That gives actual backpressure so the producer pauses instead of enqueueing unbounded work.
2. Move the permit into the task and keep it alive for the whole decode. Use OwnedSemaphorePermit so ownership survives awaits.

Cargo snippets:

# Cargo.toml
[dependencies]
anyhow = "1"
tokio = { version = "1", features = ["full"] }
tokio-util = { version = "0.7", features = ["rt"] }
tracing = "0.1"

A realistic skeleton that simulates decode work and shows the shape:

use std::{path::PathBuf, sync::Arc, time::Duration};
use anyhow::{Context, Result};
use tokio::{sync::{Semaphore}, task::JoinSet, time};

#[derive(Debug)]
struct DecodeOutput {
    path: PathBuf,
    frames: u32,
}

async fn decode_media_mock(path: PathBuf) -> Result<DecodeOutput> {
    // Simulate IO + CPU stretches
    time::sleep(Duration::from_millis(60)).await; // probing
    // pretend CPU work
    tokio::task::spawn_blocking(move || {
        // do heavy lifting: decode, scale, resample, etc.
        std::thread::sleep(Duration::from_millis(120));
        Ok::<_, anyhow::Error>(())
    }).await??;

    Ok(DecodeOutput { path, frames: 240 })
}

pub async fn run_bounded_decode(inputs: Vec<PathBuf>, max_parallel: usize) -> Result<Vec<DecodeOutput>> {
    let sem = Arc::new(Semaphore::new(max_parallel));
    let mut set = JoinSet::new();

    for path in inputs {
        // Acquire before spawning: enforces real backpressure
        let permit = sem.clone().acquire_owned().await
            .with_context(|| "semaphore closed")?;

        set.spawn(async move {
            // keep the permit alive for the entire decode
            let _permit = permit; // drop releases the slot
            decode_media_mock(path).await
        });
    }

    // Drain results in completion order
    let mut results = Vec::new();
    while let Some(joined) = set.join_next().await {
        match joined {
            Ok(Ok(out)) => results.push(out),
            Ok(Err(e)) => return Err(e).context("worker failed"), // or collect and continue
            Err(join_err) => return Err(anyhow::anyhow!(join_err)).context("task panicked/aborted"),
        }
    }

    Ok(results)
}

Notes:

• Use spawn_blocking for CPU-heavy decode to protect the async scheduler. If you only drive subprocesses like ffmpeg, you may not need it.
• Acquire before spawning so in-flight decoders never exceed max_parallel and memory stays bounded.
• JoinSet scopes the workers. If run_bounded_decode returns early or errors, the rest abort.

Dynamic cost: acquire_many for “expensive” decoders

If some decoders cost more (4K HEVC vs 720p H.264), treat the semaphore as “CPU permits” instead of “task slots.” Acquire multiple permits proportional to cost:

let cpu_permits = Arc::new(Semaphore::new(num_cpus::get() as usize));

for job in jobs {
    let cost = job.estimated_cpu_cost(); // e.g., 2 for 4K, 1 for 1080p
    let permits = cpu_permits.clone().acquire_many_owned(cost as u32).await?;
    set.spawn(async move {
        let _permits = permits;
        run_job(job).await
    });
}

This aligns runnable work with real resource use, not just a flat task count.

Pattern 2: Cancellation-safe ffmpeg workers

Subprocesses need prompt teardown on cancellation or timeouts. Combine CancellationToken, kill_on_drop, and targeted start_kill.

use anyhow::{Context, Result};
use std::{path::Path, sync::Arc, time::Duration};
use tokio::{process::Command, sync::{Semaphore}, task::JoinSet, time};
use tokio_util::sync::CancellationToken;
use tokio::io::AsyncReadExt;
use std::process::Stdio;

async fn run_ffmpeg(input: &Path, output: &Path, fps: u32, cancel: CancellationToken) -> Result<()> {
    let mut cmd = Command::new("ffmpeg");
    cmd.arg("-y")
        .args(["-hide_banner", "-loglevel", "error"]) // clean stderr
        .arg("-i").arg(input)
        .args(["-vf", &format!("fps={}", fps)])
        .arg(output)
        .stdout(Stdio::null())
        .stderr(Stdio::piped())
        .kill_on_drop(true); // if this future is dropped, the child is killed

    let mut child = cmd.spawn().context("spawn ffmpeg")?;
    let mut stderr = child.stderr.take().unwrap();

    // Pipe stderr for diagnostics while still being cancellable
    let mut stderr_buf = Vec::new();

    // Enforce an overall timeout per job
    let hard_timeout = time::sleep(Duration::from_secs(30));
    tokio::pin!(hard_timeout);

    let status = tokio::select! {
        _ = cancel.cancelled() => {
            let _ = child.start_kill();
            Err(anyhow::anyhow!("cancelled"))
        }
        _ = &mut hard_timeout => {
            let _ = child.start_kill();
            Err(anyhow::anyhow!("timeout"))
        }
        out = async {
            // Read stderr concurrently to avoid pipe filling deadlocks
            let mut read_task = tokio::spawn(async move {
                let mut local = Vec::new();
                let _ = stderr.read_to_end(&mut local).await; // ignore read errors for robustness
                local
            });
            let status = child.wait().await?;
            stderr_buf = read_task.await.unwrap_or_default();
            Ok::<_, anyhow::Error>(status)
        } => out
    }?;

    if !status.success() {
        let msg = String::from_utf8_lossy(&stderr_buf);
        anyhow::bail!("ffmpeg failed: {}", msg);
    }
    Ok(())
}

pub async fn transcode_many_ffmpeg(inputs: Vec<(String, String)>, max_parallel: usize) -> Result<()> {
    let sem = Arc::new(Semaphore::new(max_parallel));
    let cancel_all = CancellationToken::new();
    let mut set = JoinSet::new();

    for (in_path, out_path) in inputs {
        let permit = sem.clone().acquire_owned().await?;
        let token = cancel_all.child_token();
        set.spawn(async move {
            let _permit = permit; // holds the slot
            run_ffmpeg(Path::new(&in_path), Path::new(&out_path), 15, token).await
        });
    }

    // Fail-fast: cancel the batch if one task errors
    let mut first_err: Option<anyhow::Error> = None;
    while let Some(next) = set.join_next().await {
        match next {
            Ok(Ok(())) => {}
            Ok(Err(e)) => {
                if first_err.is_none() { first_err = Some(e); }
                cancel_all.cancel();
            }
            Err(join_err) => {
                if first_err.is_none() { first_err = Some(anyhow::anyhow!(join_err)); }
                cancel_all.cancel();
            }
        }
    }

    if let Some(e) = first_err { Err(e) } else { Ok(()) }
}

Key points:

• kill_on_drop(true) terminates ffmpeg if the task is aborted by dropping the JoinSet or cancelling the batch.
• start_kill() fires on timeout or token cancellation so teardown is explicit.
• CancellationToken gives group-cancel semantics: one failure cancels the rest quickly.
• Owned semaphore permits ensure we never exceed max_parallel, and the limit is held for the entire job.

Operational guidance and pitfalls

• Don’t hold permits longer than necessary. Acquire right before the work unit and drop on completion. For long but light post-processing, consider releasing early.
• Use spawn_blocking only for CPU-bound sections. If the pipeline is mostly subprocess I/O with ffmpeg, keep the async path thin.
• For heterogeneous workers, model the semaphore as resource units with acquire_many_owned instead of a flat cap.
• Prefer acquire-before-spawn to avoid a thundering herd of tasks sitting on the semaphore. Acquiring inside the task still spawns N tasks immediately.
• Attach context to errors with anyhow::Context to speed up triage when a batch fails.
• Expose metrics for queued, running, succeeded, failed, cancelled, and mean job time. They reveal mis-sized caps fast.

Why this beats ad-hoc spawn + join handles

Keeping a pile of JoinHandles and sprinkling aborts is brittle. Eventually a child leaks, error context disappears, or stderr blocks the process. The JoinSet, Semaphore, and CancellationToken trio yields:

• A single owner for the workers’ lifecycle.
• Backpressure instead of unbounded queues.
• Predictable cancellation for subprocesses and tasks.

Testing the cap and cancellation

• Set max_parallel to 1 and feed 10 inputs. Serial runtime should be roughly the sum of per-job times.
• Run with max_parallel set to N and to N+1. Expect CPU to plateau near N cores for CPU-bound work, and memory to remain stable.
• Inject a failure like a bad input path. Confirm others cancel and ffmpeg processes exit promptly.

When to reach for Rayon instead

If the hot path is purely CPU-bound Rust (no async I/O, no subprocesses), a bounded Rayon pool can give better locality and scheduling. For mixed pipelines with I/O, subprocesses, and some CPU, tokio with JoinSet and semaphores keeps the graph in one runtime with clear lifetimes.

Closing thoughts

Media decode is resource-hungry and failure-prone. With JoinSet for structured lifetimes, a semaphore for a real cap, and cancellation wired to ffmpeg, you get steady throughput, bounded memory, and clean shutdowns. Boring and correct is what you want in production.