Rust structured concurrency: cap media decoders w/ tokio JoinSet

•7 min read•Yaseen Khatib · MERN + AI Architect

Why structured concurrency for media decode

Media decode pipelines (e.g., transcoding with ffmpeg, extracting waveforms/thumbnails) invite fan-out parallelism. Without discipline, you end up with: unbounded task creation, zombie subprocesses, and nondeterministic shutdowns. Rust + tokio gives you the pieces to do this right:

  • JoinSet: spawns tasks with a scoped, structured lifetime; dropping it aborts remaining tasks.
  • Semaphore: caps concurrency and enforces backpressure.
  • CancellationToken + kill_on_drop/start_kill: cooperative, prompt teardown of child processes.

This article shows a concrete pattern to cap parallel media decoders using JoinSet and a tokio Semaphore, and how to make it cancellation-safe when wrapping ffmpeg.

JoinSet vs FuturesUnordered (and friends)

  • JoinSet spawns actual tasks. When the JoinSet is dropped, pending tasks are aborted. That’s exactly what you want for “this batch of workers belongs to this scope.”
  • FuturesUnordered runs a set of futures you provide. If you tokio::spawn inside it, you’ve detached work and must manage abort/joins yourself. If you don’t spawn, your futures must remain !Send-aware and cannot do blocking/syscall-heavy work directly.

For structured decoding workers, JoinSet is the pragmatic default.

Pattern 1: Bounded fan-out with owned semaphore permits

Two critical rules for reliable caps:

  1. Acquire a permit before spawning. That provides true backpressure; the upstream producer stalls instead of enqueuing unbounded work.
  2. Move the permit into the task and keep it alive for the whole decode. Use OwnedSemaphorePermit so ownership crosses await points safely.

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 demonstrates the structure:

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:

  • Using spawn_blocking for CPU-heavy decode is important; it protects the async scheduler. For pure subprocess-based decode (ffmpeg), you may not need it.
  • Acquire-before-spawn ensures you never have more than max_parallel in-flight decoders and your memory remains bounded.
  • JoinSet ties the workers to the scope; if run_bounded_decode returns early or errors, remaining tasks are aborted.

Dynamic cost: acquire_many for “expensive” decoders

If some decoders are heavier (e.g., 4K HEVC vs. 720p H.264), treat the semaphore as a token bucket of “CPU permits,” not “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 keeps the runnable set aligned with actual resource usage, not just a flat count of tasks.

Pattern 2: Cancellation-safe ffmpeg workers

Subprocesses must be killed promptly on cancellation/timeouts. Combine CancellationToken, kill_on_drop, and selective 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) ensures ffmpeg is terminated if the task is aborted (e.g., by dropping the JoinSet or cancelling the batch).
  • start_kill() in the select arms makes termination explicit on timeout or token cancellation.
  • The CancellationToken gives you group-cancel semantics: if any worker fails, the rest are promptly torn down.
  • Semaphore-owned permits guarantee we never exceed max_parallel processes, and the cap is held for the full duration of each job.

Operational guidance and pitfalls

  • Don’t hold permits longer than needed. Acquire right before the unit of work, and drop immediately on completion. If you need to do prolonged post-processing that isn’t resource-heavy, consider releasing early.
  • Use spawn_blocking only for CPU-bound work. If your pipeline is mostly subprocess I/O (ffmpeg), keep the async part lean.
  • If workers are heterogeneous, model the semaphore as a token bucket of “resource units” with acquire_many_owned instead of a flat concurrency cap.
  • Prefer acquire-before-spawn to avoid inflating memory and file descriptors. Acquiring inside the task still spawns N tasks immediately, all waiting on the semaphore.
  • Propagate errors with context (anyhow::Context) to make triage easier when a batch fails.
  • Metrics matter: expose counters for queued, running, succeeded, failed, cancelled, and average job time. It’s the fastest way to spot mis-sized caps.

Why this beats ad-hoc spawn + join handles

Manually tracking JoinHandles and relying on ad-hoc aborts is fragile: you’ll eventually leak a child, lose error provenance, or deadlock stderr. The JoinSet + Semaphore + CancellationToken trio gives you:

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

Testing the cap and cancellation

  • Set max_parallel to 1 and feed 10 inputs; verify serial execution time ≈ sum of per-job times.
  • Set max_parallel to N and to N+1; ensure CPU saturates at about N cores (if CPU-bound) and memory doesn’t spike.
  • Inject a failure (e.g., bad input path) and confirm all other tasks cancel and ffmpeg processes die promptly.

When to reach for Rayon instead

If the hot path is purely CPU-bound Rust code (no async I/O, no subprocesses), Rayon with a bounded thread pool may deliver better locality and scheduling. For mixed pipelines (I/O + subprocess + some CPU), tokio with JoinSet + semaphores keeps the whole graph in one runtime with structured lifetimes.

Closing thoughts

Media decode is resource-hungry and failure-prone. With JoinSet providing structured lifetimes, a semaphore enforcing a real cap, and cancellation wired through to ffmpeg, you get predictable throughput, bounded memory, and clean shutdowns. That’s the kind of boring, correct concurrency you want in production.

Looking to architect a similar system?

Let's ship it at AI-speed.

Start a conversation →