Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Introduction

Ichika is a Rust procedural macro library for building thread pool based pipelines with automatic error handling, retry semantics, and graceful shutdown support.

Overview

Ichika provides a powerful pipe! macro that allows you to define complex multi-stage processing pipelines where each stage runs in its own thread pool. The macro handles all the boilerplate of creating thread pools, setting up communication channels, and coordinating between stages.

Key Features

  • Declarative Pipeline Syntax: Define complex processing pipelines using a clean, expressive macro syntax
  • Automatic Thread Pool Management: Each stage gets its own dedicated thread pool
  • Error Propagation: Built-in error handling with Result types throughout the pipeline
  • Retry Semantics: Configurable retry policies for handling transient failures
  • Async Runtime Agnostic: Works with both tokio and async-std
  • Graceful Shutdown: Proper cleanup when the pipeline is dropped
  • Monitoring: Built-in thread usage statistics and task counting

A Simple Example

use ichika::prelude::*;

fn main() -> anyhow::Result<()> {
    // Create a simple 2-stage pipeline
    let pool = pipe![
        |req: String| -> usize {
            Ok(req.len())
        },
        |req: usize| -> String {
            Ok(req.to_string())
        }
    ]?;

    // Send some requests
    pool.send("hello".to_string())?;
    pool.send("world".to_string())?;

    // Collect results
    while let Some(result) = pool.recv()? {
        println!("Got: {}", result);
    }

    Ok(())
}

Use Cases

Ichika is particularly useful for:

  • Data Processing Pipelines: Multi-stage data transformation workflows
  • API Request Handling: Processing requests through multiple validation/transformation stages
  • Event Processing: Building event-driven systems with staged processing
  • Batch Jobs: Parallel processing with configurable concurrency per stage
  • Microservices: Internal service communication with bounded queues

Design Philosophy

Ichika follows these principles:

  1. Safety First: Leverage Rust’s type system for compile-time guarantees
  2. Ergonomic API: Minimize boilerplate while maintaining flexibility
  3. Zero Cost Abstractions: No runtime overhead beyond what’s necessary
  4. Explicit Control: Give users fine-grained control over thread pools and queues

Project Status

Ichika is currently in active development. The API may change between versions, but we strive to maintain backward compatibility whenever possible.

License

Ichika is licensed under the MIT License. See LICENSE for details.

Getting Started

This guide will help you get started with Ichika, from installation to your first pipeline.

Installation

Add Ichika to your Cargo.toml:

[dependencies]
ichika = "0.1"

Feature Flags

Ichika supports different async runtimes via feature flags:

# For tokio support (default)
ichika = { version = "0.1", features = ["tokio"] }

# For async-std support
ichika = { version = "0.1", features = ["async-std"] }

# For both runtimes
ichika = { version = "0.1", features = ["tokio", "async-std"] }

Your First Pipeline

Let’s create a simple pipeline that processes strings:

use ichika::prelude::*;

fn main() -> anyhow::Result<()> {
    // Define a 3-stage pipeline
    let pool = pipe![
        // Stage 1: Parse string to number
        |req: String| -> anyhow::Result<usize> {
            req.parse::<usize>()
                .map_err(|e| anyhow::anyhow!("Failed to parse: {}", e))
        },
        // Stage 2: Double the number
        |req: anyhow::Result<usize>| -> anyhow::Result<usize> {
            req.map(|n| n * 2)
        },
        // Stage 3: Convert back to string
        |req: anyhow::Result<usize>| -> String {
            req.map(|n| n.to_string())
                .unwrap_or_else(|e| format!("Error: {}", e))
        }
    ]?;

    // Process some data
    pool.send("42".to_string())?;
    pool.send("100".to_string())?;
    pool.send("invalid".to_string())?;

    // Collect results
    for _ in 0..3 {
        if let Some(result) = pool.recv()? {
            println!("Result: {}", result);
        }
    }

    Ok(())
}

Understanding the Basics

The pipe! Macro

The pipe! macro creates a chain of processing stages. Each stage:

  1. Receives input from the previous stage (or the initial send() call)
  2. Processes the data in a thread pool
  3. Passes the result to the next stage

Type Propagation

Ichika automatically infers the types flowing through your pipeline:

#![allow(unused)]
fn main() {
let pool = pipe![
    |req: String| -> usize { req.len() },     // String -> usize
    |req: usize| -> String { req.to_string() } // usize -> String
]?;
}

Error Handling

Each stage can return a Result, and errors are automatically propagated:

#![allow(unused)]
fn main() {
let pool = pipe![
    |req: String| -> anyhow::Result<i32> {
        req.parse().map_err(Into::into)
    },
    |req: anyhow::Result<i32>| -> i32 {
        req.unwrap() // or handle the error appropriately
    }
]?;
}

Next Steps

The pipe! Macro

The pipe! macro is the core of Ichika. It transforms a sequence of closures into a fully-functional multi-stage processing pipeline.

Basic Syntax

#![allow(unused)]
fn main() {
let pool = pipe![
    closure1,
    closure2,
    closure3,
    // ... more closures
]?;
}

Each closure represents one processing stage in your pipeline.

Closure Signatures

Each closure must follow these rules:

  1. Accept exactly one parameter - the input from the previous stage
  2. Return a type - this becomes the input to the next stage
  3. Be Clone + Send + 'static - required for thread pool execution

Example Signatures

#![allow(unused)]
fn main() {
|req: String| -> usize {
    req.len()
}

|req: usize| -> anyhow::Result<String> {
    Ok(req.to_string())
}

|req: anyhow::Result<MyData>| -> MyOutput {
    // Handle the Result
}
}

Type Inference

Ichika automatically connects the output type of one stage to the input type of the next:

#![allow(unused)]
fn main() {
let pool = pipe![
    |req: String| -> usize {        // Stage 1: String -> usize
        req.len()
    },
    |req: usize| -> String {         // Stage 2: usize -> String
        req.to_string()
    },
    |req: String| -> bool {          // Stage 3: String -> bool
        !req.is_empty()
    }
]?;
}

Stage Attributes

You can configure individual stages using attributes:

Thread Pool Configuration

#![allow(unused)]
fn main() {
let pool = pipe![
    #[threads(4)]                    // Use 4 threads for this stage
    |req: String| -> usize {
        req.len()
    },
    #[threads(2)]                    // Use 2 threads for this stage
    |req: usize| -> String {
        req.to_string()
    }
]?;
}

Queue Configuration

#![allow(unused)]
fn main() {
let pool = pipe![
    #[queue(100)]                    // Queue capacity of 100
    |req: String| -> usize {
        req.len()
    }
]?;
}

Named Stages

#![allow(unused)]
fn main() {
let pool = pipe![
    #[name("parser")]                // Name the stage for monitoring
    |req: String| -> usize {
        req.len()
    },
    #[name("formatter")]
    |req: usize| -> String {
        req.to_string()
    }
]?;

// Query task count for a named stage
let count = pool.task_count("parser")?;
}

Branching Pipelines

You can create conditional branching in your pipeline:

#![allow(unused)]
fn main() {
let pool = pipe![
    |req: String| -> anyhow::Result<Either<usize, String>> {
        if req.parse::<usize>().is_ok() {
            Ok(Either::Left(req.parse::<usize>()?))
        } else {
            Ok(Either::Right(req))
        }
    },
    // Handle each branch
    |req: Either<usize, String>| -> String {
        match req {
            Either::Left(n) => format!("Number: {}", n),
            Either::Right(s) => format!("String: {}", s),
        }
    }
]?;
}

Async Stages

With the appropriate feature flag, you can use async stages:

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let pool = pipe![
        |req: String| -> usize {
            req.len()
        },
        async |req: usize| -> String {
            // This runs in the async runtime
            tokio::time::sleep(Duration::from_millis(100)).await;
            req.to_string()
        }
    ]?;

    Ok(())
}

Global Constraints

You can set global constraints for the entire pipeline:

#![allow(unused)]
fn main() {
let pool = pipe![
    #[global_threads(8)]             // Default thread count for all stages
    #[global_queue(1000)]            // Default queue capacity
    |req: String| -> usize {
        req.len()
    },
    |req: usize| -> String {
        req.to_string()
    }
]?;
}

Complete Example

Here’s a more realistic example showing multiple features:

use ichika::prelude::*;

fn main() -> anyhow::Result<()> {
    env_logger::init();

    let pool = pipe![
        #[name("parse")]
        #[threads(2)]
        |req: String| -> anyhow::Result<i32> {
            log::info!("Parsing: {}", req);
            req.parse().map_err(Into::into)
        },
        #[name("process")]
        #[threads(4)]
        |req: anyhow::Result<i32>| -> anyhow::Result<i32> {
            let n = req?;
            log::info!("Processing: {}", n);
            Ok(n * 2)
        },
        #[name("format")]
        |req: anyhow::Result<i32>| -> String {
            match req {
                Ok(n) => {
                    log::info!("Formatting: {}", n);
                    format!("Result: {}", n)
                }
                Err(e) => {
                    log::error!("Error: {}", e);
                    format!("Error: {}", e)
                }
            }
        }
    ]?;

    // Monitor thread usage
    println!("Thread usage: {}", pool.thread_usage()?);

    Ok(())
}

ThreadPool Trait

The ThreadPool trait defines the interface for all pipeline pools created by the pipe! macro.

Trait Definition

#![allow(unused)]
fn main() {
pub trait ThreadPool {
    type Request: Clone;
    type Response: Clone;

    fn send(&self, req: Self::Request) -> Result<()>;
    fn recv(&self) -> Result<Option<Self::Response>>;

    fn thread_usage(&self) -> Result<usize>;
    fn task_count(&self, id: impl ToString) -> Result<usize>;
}
}

Methods

send

Sends a request to the pipeline for processing.

#![allow(unused)]
fn main() {
fn send(&self, req: Self::Request) -> Result<()>
}

Parameters:

  • req - The request to send, must match the pipeline’s input type

Returns:

  • Result<()> - Ok if successfully queued, Err if the send fails

Example:

#![allow(unused)]
fn main() {
let pool = pipe![
    |req: String| -> usize { req.len() }
]?;

pool.send("hello".to_string())?;
}

recv

Receives the next processed result from the pipeline.

#![allow(unused)]
fn main() {
fn recv(&self) -> Result<Option<Self::Response>>
}

Returns:

  • Ok(Some(response)) - A processed result
  • Ok(None) - The pipeline has terminated
  • Err(...) - An error occurred while receiving

Example:

#![allow(unused)]
fn main() {
loop {
    match pool.recv()? {
        Some(result) => println!("Got: {}", result),
        None => break,
    }
}
}

thread_usage

Returns the current number of threads in use by the pipeline.

#![allow(unused)]
fn main() {
fn thread_usage(&self) -> Result<usize>
}

Returns:

  • The total number of active threads across all stages

Example:

#![allow(unused)]
fn main() {
println!("Active threads: {}", pool.thread_usage()?);
}

task_count

Returns the number of pending tasks for a named stage.

#![allow(unused)]
fn main() {
fn task_count(&self, id: impl ToString) -> Result<usize>
}

Parameters:

  • id - The stage name (as set by #[name(...)] attribute)

Returns:

  • The number of tasks waiting in that stage’s queue

Example:

#![allow(unused)]
fn main() {
let pool = pipe![
    #[name("parser")]
    |req: String| -> usize { req.len() }
]?;

pool.send("test".to_string())?;
println!("Parser queue depth: {}", pool.task_count("parser")?);
}

Type Parameters

Request

The input type for the pipeline. This is the type accepted by the first stage.

#![allow(unused)]
fn main() {
let pool: impl ThreadPool<Request = String, Response = usize> = pipe![
    |req: String| -> usize { req.len() }
]?;
}

Response

The output type of the pipeline. This is the type returned by the last stage.

#![allow(unused)]
fn main() {
let pool: impl ThreadPool<Request = String, Response = String> = pipe![
    |req: String| -> usize { req.len() },
    |req: usize| -> String { req.to_string() }
]?;
}

Lifecycle

The pipeline follows this lifecycle:

  1. Created - The pipe! macro returns a new pool
  2. Active - You can send() requests and recv() results
  3. Draining - When dropped, the pool finishes processing pending tasks
  4. Terminated - recv() returns None when the pool is shut down

Graceful Shutdown

When the pool is dropped, it:

  1. Stops accepting new requests
  2. Finishes processing all queued tasks
  3. Shuts down all thread pools gracefully
#![allow(unused)]
fn main() {
{
    let pool = pipe![
        |req: String| -> usize { req.len() }
    ]?;

    pool.send("hello".to_string())?;
    // pool goes out of scope and shuts down gracefully
}
}

Monitoring

Use the monitoring methods to track pipeline health:

use ichika::prelude::*;
use std::time::Duration;

fn main() -> anyhow::Result<()> {
    let pool = pipe![
        #[name("stage1")]
        |req: String| -> usize { req.len() },
        #[name("stage2")]
        |req: usize| -> String { req.to_string() }
    ]?;

    // Send work
    for i in 0..100 {
        pool.send(format!("request-{}", i))?;
    }

    // Monitor progress
    loop {
        let threads = pool.thread_usage()?;
        let stage1_pending = pool.task_count("stage1")?;
        let stage2_pending = pool.task_count("stage2")?;

        println!(
            "Threads: {}, Stage1 pending: {}, Stage2 pending: {}",
            threads, stage1_pending, stage2_pending
        );

        if stage1_pending == 0 && stage2_pending == 0 {
            break;
        }

        std::thread::sleep(Duration::from_millis(100));
    }

    Ok(())
}

Error Handling & Retry

Ichika provides robust error handling with built-in retry semantics for transient failures.

Error Propagation

Errors naturally flow through the pipeline using Result types:

#![allow(unused)]
fn main() {
let pool = pipe![
    |req: String| -> anyhow::Result<i32> {
        req.parse().map_err(Into::into)
    },
    |req: anyhow::Result<i32>| -> anyhow::Result<i32> {
        let n = req?;
        Ok(n * 2)
    },
    |req: anyhow::Result<i32>| -> String {
        match req {
            Ok(n) => format!("Result: {}", n),
            Err(e) => format!("Error: {}", e),
        }
    }
]?;
}

Type Transformation

When a stage returns a Result, the next stage receives that Result:

#![allow(unused)]
fn main() {
|req: String| -> anyhow::Result<usize> { ... }  // Returns Result
|req: anyhow::Result<usize>| -> usize {         // Receives Result
    req.unwrap()
}
}

Retry Semantics

Ichika provides automatic retry for operations that may fail transiently.

Basic Retry

Use the retry function to retry an operation:

#![allow(unused)]
fn main() {
use ichika::retry;

let result = retry(|| {
    // Operation that might fail
    Ok::<_, anyhow::Error>(42)
})?;
}

Retry with Policy

Control retry behavior with a RetryPolicy:

#![allow(unused)]
fn main() {
use ichika::{retry_with, RetryPolicy};
use std::time::Duration;

let policy = RetryPolicy {
    max_attempts: 3,
    backoff: Duration::from_millis(100),
    ..Default::default()
};

let result = retry_with(policy, || {
    // Operation with custom retry policy
    Ok::<_, anyhow::Error>(42)
})?;
}

RetryPolicy Options

#![allow(unused)]
fn main() {
pub struct RetryPolicy {
    /// Maximum number of retry attempts
    pub max_attempts: usize,

    /// Initial backoff duration (exponential backoff is applied)
    pub backoff: Duration,

    /// Maximum backoff duration
    pub max_backoff: Duration,

    /// Whether to use jitter in backoff calculation
    pub jitter: bool,
}

impl Default for RetryPolicy {
    fn default() -> Self {
        Self {
            max_attempts: 3,
            backoff: Duration::from_millis(100),
            max_backoff: Duration::from_secs(30),
            jitter: true,
        }
    }
}
}

Using Retry in Pipelines

Retry Within a Stage

#![allow(unused)]
fn main() {
let pool = pipe![
    #[name("fetch")]
    |req: String| -> anyhow::Result<String> {
        // Retry the fetch operation
        retry_with(
            RetryPolicy {
                max_attempts: 3,
                backoff: Duration::from_millis(100),
                ..Default::default()
            },
            || {
                // Simulated fetch that might fail
                if rand::random::<f32>() < 0.3 {
                    Err(anyhow::anyhow!("Network error"))
                } else {
                    Ok(format!("Fetched: {}", req))
                }
            }
        )
    }
]?;
}

Retry at Pipeline Level

For more control, handle retry at the caller level:

#![allow(unused)]
fn main() {
fn process_with_retry(pool: &impl ThreadPool<Request = String, Response = String>, input: String) -> anyhow::Result<String> {
    retry_with(
        RetryPolicy {
            max_attempts: 5,
            backoff: Duration::from_millis(50),
            ..Default::default()
        },
        || {
            pool.send(input.clone())?;
            match pool.recv()? {
                Some(result) => Ok(result),
                None => Err(anyhow::anyhow!("Pipeline terminated")),
            }
        }
    )
}
}

Error Recovery Strategies

Fallback Values

#![allow(unused)]
fn main() {
let pool = pipe![
    |req: String| -> anyhow::Result<i32> {
        req.parse().map_err(Into::into)
    },
    |req: anyhow::Result<i32>| -> i32 {
        req.unwrap_or(0)  // Default to 0 on error
    }
]?;
}

Error Aggregation

#![allow(unused)]
fn main() {
let pool = pipe![
    |req: Vec<String>| -> Vec<anyhow::Result<i32>> {
        req.into_iter()
            .map(|s| s.parse::<i32>().map_err(Into::into))
            .collect()
    },
    |req: Vec<anyhow::Result<i32>>| -> (i32, usize) {
        let (sum, errors) = req.into_iter().fold(
            (0, 0),
            |(sum, errs), r| match r {
                Ok(n) => (sum + n, errs),
                Err(_) => (sum, errs + 1),
            },
        );
        (sum, errors)
    }
]?;
}

Circuit Breaker Pattern

#![allow(unused)]
fn main() {
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;

let circuit_breaker = Arc::new(AtomicBool::new(false));

let pool = pipe![
    |req: String| -> anyhow::Result<String> {
        if circuit_breaker.load(Ordering::Relaxed) {
            return Err(anyhow::anyhow!("Circuit breaker is open"));
        }
        // Process request
        Ok(format!("Processed: {}", req))
    }
]?;
}

Complete Example

Here’s a comprehensive example showing error handling with retry:

use ichika::prelude::*;
use ichika::{retry_with, RetryPolicy};
use std::time::Duration;

fn main() -> anyhow::Result<()> {
    let pool = pipe![
        #[name("validate")]
        |req: String| -> anyhow::Result<i32> {
            req.parse()
                .map_err(|e| anyhow::anyhow!("Invalid input: {}", e))
        },
        #[name("process")]
        |req: anyhow::Result<i32>| -> anyhow::Result<i32> {
            let n = req?;
            // Simulate transient failure
            if n % 3 == 0 {
                Err(anyhow::anyhow!("Transient error"))
            } else {
                Ok(n * 2)
            }
        },
        #[name("format")]
        |req: anyhow::Result<i32>| -> String {
            match req {
                Ok(n) => format!("Success: {}", n),
                Err(e) => format!("Failed: {}", e),
            }
        }
    ]?;

    // Send various inputs
    let inputs = vec!["10", "20", "30", "invalid", "40"];

    for input in inputs {
        pool.send(input.to_string())?;
    }

    // Collect results
    loop {
        match pool.recv()? {
            Some(result) => println!("{}", result),
            None => break,
        }
    }

    Ok(())
}

Best Practices

  1. Use anyhow::Result for flexible error handling
  2. Set appropriate retry limits to avoid infinite loops
  3. Use exponential backoff for network operations
  4. Log errors appropriately for debugging
  5. Consider circuit breakers for external service calls
  6. Make errors informative - include context about what failed

Advanced Features

This section covers advanced features and techniques for getting the most out of Ichika.

Async Integration

Ichika supports both tokio and async-std runtimes. Enable with feature flags:

[dependencies]
ichika = { version = "0.1", features = ["tokio"] }
# or
ichika = { version = "0.1", features = ["async-std"] }

Async Stages

Mix sync and async stages seamlessly:

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let pool = pipe![
        |req: String| -> usize {
            req.len()  // Sync stage
        },
        async |req: usize| -> String {
            // Async stage - runs in tokio runtime
            tokio::time::sleep(Duration::from_millis(100)).await;
            req.to_string()
        }
    ]?;

    Ok(())
}

Custom Thread Creators

You can customize how threads are created for each stage:

#![allow(unused)]
fn main() {
use std::thread;

let pool = pipe![
    #[creator(|name| {
        thread::Builder::new()
            .name(name.to_string())
            .stack_size(2 * 1024 * 1024)  // 2MB stack
            .spawn(|| {
                // Custom thread logic
            })
    })]
    |req: String| -> usize {
        req.len()
    }
]?;
}

Monitoring and Observability

Thread Usage Tracking

#![allow(unused)]
fn main() {
let pool = pipe![
    #[name("worker")]
    |req: String| -> usize {
        req.len()
    }
]?;

// Get total thread count
let total_threads = pool.thread_usage()?;

// Get pending tasks for a named stage
let pending = pool.task_count("worker")?;

println!("Threads: {}, Pending: {}", total_threads, pending);
}

Health Checks

#![allow(unused)]
fn main() {
fn check_pool_health(pool: &impl ThreadPool) -> anyhow::Result<bool> {
    let threads = pool.thread_usage()?;
    let is_healthy = threads > 0;
    Ok(is_healthy)
}
}

Resource Management

Graceful Shutdown

#![allow(unused)]
fn main() {
use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};

let running = Arc::new(AtomicBool::new(true));
let r = running.clone();

// Spawn a monitoring thread
thread::spawn(move || {
    while r.load(Ordering::Relaxed) {
        // Monitor pool health
        thread::sleep(Duration::from_secs(1));
    }
});

// When done, set running to false
running.store(false, Ordering::Relaxed);
// Pool will shut down gracefully when dropped
}

Memory Considerations

Each stage has a bounded queue. Adjust queue sizes based on your memory constraints:

#![allow(unused)]
fn main() {
let pool = pipe![
    #[queue(100)]   // Small queue for memory-constrained environments
    |req: String| -> usize {
        req.len()
    },
    #[queue(1000)]  // Larger queue for high-throughput stages
    |req: usize| -> String {
        req.to_string()
    }
]?;
}

Pipeline Patterns

Fan-Out / Fan-In

Process items in parallel and collect results:

#![allow(unused)]
fn main() {
let pool = pipe![
    |req: Vec<String>| -> Vec<String> {
        req.into_iter()
            .filter(|s| !s.is_empty())
            .collect()
    },
    |req: Vec<String>| -> usize {
        req.len()
    }
]?;
}

Stateful Processing

Use Arc<Mutex<T>> for stateful stages:

#![allow(unused)]
fn main() {
use std::sync::{Arc, Mutex};

let counter = Arc::new(Mutex::new(0));
let c = counter.clone();

let pool = pipe![
    move |req: String| -> usize {
        let mut count = c.lock().unwrap();
        *count += 1;
        println!("Processed {} items", *count);
        req.len()
    }
]?;
}

Conditional Routing

#![allow(unused)]
fn main() {
enum Event {
    Login(String),
    Logout(String),
    Message(String, String),
}

let pool = pipe![
    |req: Event| -> String {
        match req {
            Event::Login(user) => format!("Login: {}", user),
            Event::Logout(user) => format!("Logout: {}", user),
            Event::Message(from, msg) => format!("{}: {}", from, msg),
        }
    }
]?;
}

Performance Tuning

Thread Pool Sizing

#![allow(unused)]
fn main() {
let num_cpus = num_cpus::get();

let pool = pipe![
    #[threads(num_cpus)]  // Match CPU count
    |req: String| -> usize {
        req.len()
    }
]?;
}

Batch Processing

#![allow(unused)]
fn main() {
let pool = pipe![
    |req: Vec<String>| -> Vec<usize> {
        req.par_iter()  // Use rayon for parallel processing
            .map(|s| s.len())
            .collect()
    }
]?;
}

Testing Pipelines

Unit Testing Stages

#![allow(unused)]
fn main() {
#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_pipeline() {
        let pool = pipe![
            |req: String| -> usize { req.len() },
            |req: usize| -> String { req.to_string() }
        ].unwrap();

        pool.send("test".to_string()).unwrap();
        let result = pool.recv().unwrap().unwrap();
        assert_eq!(result, "4");
    }
}
}

Integration Testing

#![allow(unused)]
fn main() {
#[test]
fn test_error_handling() {
    let pool = pipe![
        |req: String| -> anyhow::Result<i32> {
            req.parse().map_err(Into::into)
        }
    ].unwrap();

    pool.send("invalid".to_string()).unwrap();
    // Pipeline should handle errors gracefully
}
}

Best Practices

  1. Name your stages for better monitoring and debugging
  2. Use appropriate thread counts - don’t oversubscribe your CPU
  3. Set reasonable queue sizes to bound memory usage
  4. Handle errors explicitly - don’t silently ignore failures
  5. Monitor resource usage in production
  6. Test error paths - not just happy paths
  7. Consider backpressure - what happens when downstream is slow?
  8. Use async for I/O-bound stages, sync for CPU-bound stages

Examples

This page contains practical examples demonstrating various Ichika features.

Table of Contents

Basic Synchronous Pipeline

A minimal example showing a simple 2-stage synchronous pipeline:

use ichika::prelude::*;

fn main() -> anyhow::Result<()> {
    env_logger::builder()
        .filter_level(log::LevelFilter::Info)
        .init();

    let pool = pipe![
        |req: String| -> usize {
            log::info!("Converting '{}' to length", req);
            Ok(req.len())
        },
        |req: usize| -> String {
            log::info!("Converting length {} back to string", req);
            Ok(req.to_string())
        }
    ]?;

    let inputs = vec!["hello", "world", "ichika"];

    for input in inputs {
        pool.send(input.to_string())?;
    }

    std::thread::sleep(std::time::Duration::from_millis(500));

    loop {
        match pool.recv()? {
            Some(output) => log::info!("Received: {}", output),
            None => break,
        }
    }

    Ok(())
}

Basic Asynchronous Pipeline

Example using async stages with tokio:

use ichika::prelude::*;
use std::time::Duration;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    env_logger::init();

    let pool = pipe![
        |req: String| -> usize {
            log::info!("Stage 1: {}", req);
            req.len()
        },
        async |req: usize| -> String {
            log::info!("Stage 2: processing {}", req);
            tokio::time::sleep(Duration::from_millis(100)).await;
            req.to_string()
        }
    ]?;

    pool.send("async".to_string())?;
    pool.send("pipeline".to_string())?;

    tokio::time::sleep(Duration::from_secs(1)).await;

    loop {
        match pool.recv()? {
            Some(result) => println!("Result: {}", result),
            None => break,
        }
    }

    Ok(())
}

Error Handling

Demonstrating error propagation through the pipeline:

use ichika::prelude::*;

fn main() -> anyhow::Result<()> {
    let pool = pipe![
        #[name("parse")]
        |req: String| -> anyhow::Result<i32> {
            log::info!("Parsing: {}", req);
            req.parse().map_err(Into::into)
        },
        #[name("process")]
        |req: anyhow::Result<i32>| -> anyhow::Result<i32> {
            let n = req?;
            log::info!("Processing: {}", n);
            Ok(n * 2)
        },
        #[name("format")]
        |req: anyhow::Result<i32>| -> String {
            match req {
                Ok(n) => format!("Result: {}", n),
                Err(e) => format!("Error: {}", e),
            }
        }
    ]?;

    let inputs = vec!["42", "100", "invalid", "200"];

    for input in inputs {
        pool.send(input.to_string())?;
    }

    std::thread::sleep(std::time::Duration::from_millis(100));

    loop {
        match pool.recv()? {
            Some(result) => println!("{}", result),
            None => break,
        }
    }

    Ok(())
}

Graceful Shutdown

Demonstrating proper cleanup when the pipeline is dropped:

use ichika::prelude::*;
use std::time::Duration;

fn main() -> anyhow::Result<()> {
    env_logger::init();

    {
        let pool = pipe![
            |req: String| -> usize {
                log::info!("Processing: {}", req);
                std::thread::sleep(Duration::from_millis(50));
                req.len()
            }
        ]?;

        // Send work
        for i in 0..10 {
            pool.send(format!("request-{}", i))?;
        }

        // Give some time for processing
        std::thread::sleep(Duration::from_millis(200));

        // Pool will shut down gracefully when dropped
        log::info!("Pool going out of scope...");
    }

    log::info!("Pool has shut down gracefully");

    Ok(())
}

Monitoring Thread Usage

Track thread usage and task counts:

use ichika::prelude::*;
use std::time::Duration;

fn main() -> anyhow::Result<()> {
    let pool = pipe![
        #[name("stage1")]
        |req: String| -> usize {
            std::thread::sleep(Duration::from_millis(100));
            req.len()
        },
        #[name("stage2")]
        |req: usize| -> String {
            req.to_string()
        }
    ]?;

    // Send some work
    for i in 0..50 {
        pool.send(format!("request-{}", i))?;
    }

    // Monitor progress
    loop {
        let threads = pool.thread_usage()?;
        let stage1_pending = pool.task_count("stage1")?;
        let stage2_pending = pool.task_count("stage2")?;

        println!(
            "Threads: {}, Stage1: {}, Stage2: {}",
            threads, stage1_pending, stage2_pending
        );

        if stage1_pending == 0 && stage2_pending == 0 {
            break;
        }

        std::thread::sleep(Duration::from_millis(100));
    }

    println!("All tasks completed");

    Ok(())
}

Tuple Payload Pipeline

Working with tuple payloads:

use ichika::prelude::*;

fn main() -> anyhow::Result<()> {
    let pool = pipe![
        |req: String| -> (String, usize) {
            let len = req.len();
            (req, len)
        },
        |req: (String, usize)| -> String {
            format!("'{}' has length {}", req.0, req.1)
        }
    ]?;

    pool.send("hello".to_string())?;
    pool.send("world".to_string())?;

    std::thread::sleep(std::time::Duration::from_millis(100));

    loop {
        match pool.recv()? {
            Some(result) => println!("{}", result),
            None => break,
        }
    }

    Ok(())
}

Running the Examples

All examples are available in the repository:

# Run a specific example
cargo run --example basic_sync_chain

# Run with logging
RUST_LOG=info cargo run --example basic_sync_chain

# Run async example
cargo run --example basic_async_chain --features tokio

More Examples

Check the examples/ directory in the repository for more complete examples:

  • basic_sync_chain.rs - Synchronous pipeline
  • basic_async_chain.rs - Asynchronous pipeline
  • error_handling.rs - Error propagation
  • graceful_shutdown_drop.rs - Cleanup on drop
  • monitoring_thread_usage.rs - Monitoring APIs
  • tuple_payload_pipeline.rs - Complex payload types
  • status_exit_demo.rs - Status and exit handling