Crate noworkers 

noworkers

A small, ergonomic Rust crate for spawning and supervising groups of asynchronous “workers” on Tokio. Manage concurrent tasks with optional limits, cancellation, and first-error propagation.

This crate is inspired by Go’s errgroup package, providing similar functionality in an idiomatic Rust way.

Overview

noworkers provides a simple way to manage groups of concurrent async tasks with:

• Bounded or unbounded concurrency - Control how many tasks run simultaneously
• Automatic cancellation - First error cancels all remaining tasks
• Flexible cancellation strategies - External tokens or task-driven cancellation
• Zero-cost abstractions - Minimal overhead over raw tokio tasks

Quick Start

use noworkers::Workers;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    // Create a new worker group
    let mut workers = Workers::new();
     
    // Limit concurrent tasks to 5
    workers.with_limit(5);
     
    // Spawn 10 tasks
    for i in 0..10 {
        workers.add(move |_cancel| async move {
            println!("Task {i} running");
            tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
            Ok(())
        }).await?;
    }
     
    // Wait for all tasks to complete
    workers.wait().await?;
    Ok(())
}

Core Concepts

Worker Groups

A Workers instance represents a group of related async tasks that should be managed together. All tasks in a group share:

• A common concurrency limit (if set)
• A cancellation token hierarchy
• First-error propagation semantics

Error Handling

The first task to return an error “wins” - its error is captured and all other tasks are immediately cancelled. This provides fail-fast semantics similar to Go’s errgroup.

use noworkers::Workers;

let workers = Workers::new();

// This task will fail first
workers.add(|_| async {
    tokio::time::sleep(tokio::time::Duration::from_millis(100)).await;
    Err(anyhow::anyhow!("task failed"))
}).await?;

// These tasks will be cancelled
for i in 0..5 {
    workers.add(move |cancel| async move {
        // Check for cancellation
        tokio::select! {
            _ = tokio::time::sleep(tokio::time::Duration::from_secs(10)) => {
                Ok(())
            }
            _ = cancel.cancelled() => {
                println!("Task {i} cancelled");
                Ok(())
            }
        }
    }).await?;
}

// This will return the error from the first task
let result = workers.wait().await;
assert!(result.is_err());

Concurrency Limits

You can limit how many tasks run concurrently using Workers::with_limit. When the limit is reached, new tasks will wait for a slot to become available.

use noworkers::Workers;
use std::sync::Arc;
use std::sync::atomic::{AtomicUsize, Ordering};

let mut workers = Workers::new();
workers.with_limit(2); // Only 2 tasks run at once

let concurrent_count = Arc::new(AtomicUsize::new(0));

for i in 0..10 {
    let count = concurrent_count.clone();
    workers.add(move |_| async move {
        let current = count.fetch_add(1, Ordering::SeqCst) + 1;
        assert!(current <= 2, "Too many concurrent tasks!");
         
        tokio::time::sleep(tokio::time::Duration::from_millis(10)).await;
         
        count.fetch_sub(1, Ordering::SeqCst);
        Ok(())
    }).await?;
}

workers.wait().await?;

Cancellation

There are two ways to trigger cancellation:

1. External cancellation - Use an existing CancellationToken
2. Task-driven cancellation - A dedicated task that triggers cancellation when complete

use noworkers::Workers;
use tokio_util::sync::CancellationToken;

// External cancellation
let mut workers = Workers::new();
let cancel = CancellationToken::new();
workers.with_cancel(&cancel);

// Cancel after 1 second
let cancel_clone = cancel.clone();
tokio::spawn(async move {
    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
    cancel_clone.cancel();
});

// Task-driven cancellation
let mut workers2 = Workers::new();
workers2.with_cancel_task(async {
    tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
});

Extensions

The crate provides extension traits for common patterns:

use noworkers::Workers;
use noworkers::extensions::WithSysLimitCpus;

let mut workers = Workers::new();
// Automatically limit to number of CPU cores
workers.with_limit_to_system_cpus();

Examples

See the examples/ directory for more complete examples:

• basic.rs - Simple task spawning and waiting
• with_limit.rs - Concurrency limiting
• cancellation.rs - Cancellation patterns
• error_handling.rs - Error propagation
• web_scraper.rs - Real-world web scraping example
• parallel_processing.rs - Data processing pipeline

Modules

extensions

Extension traits for common patterns.

Structs

WorkerGuard

Guard that tracks active worker slots for concurrency limiting.

Workers

A group of supervised async workers with optional concurrency limits.