Async Rust for Security Tools: Tokio and Concurrency (2026)

Master async programming in Rust for security tools. Learn Tokio fundamentals, concurrent processing patterns, I/O optimization, and production-ready async techniques for building high-performance security applications.

Key Takeaways

• Tokio Fundamentals: Understand async runtime and futures
• Concurrent Processing: Handle multiple tasks efficiently
• I/O Optimization: Async network and file operations
• Error Handling: Async error handling patterns
• Production Patterns: Best practices for async security tools
• Performance: Optimize async code for security workloads

Table of Contents

1. Why Async for Security Tools
2. Tokio Fundamentals
3. Concurrent Processing
4. Async I/O Patterns
5. Error Handling
6. Advanced Patterns
7. Troubleshooting Guide
8. Real-World Case Study
9. FAQ
10. Conclusion

TL;DR

Master async Rust with Tokio for security tools. Learn concurrent processing, async I/O, and production patterns for building high-performance security applications.

Prerequisites

• Rust 1.80+ installed
• Understanding of Rust basics
• Familiarity with async concepts (helpful but not required)

Safety and Legal

• Use async patterns for authorized tools only
• Test concurrent code thoroughly
• Ensure thread safety in shared state
• Document async patterns clearly

Why Async for Security Tools

Performance Benefits

Concurrent I/O:

• Process multiple network connections simultaneously
• Handle thousands of concurrent operations
• Efficient resource utilization

Scalability:

• Handle high connection counts
• Low memory overhead per task
• Better CPU utilization

Tokio Fundamentals

Tokio Runtime Configuration

Understanding Runtime Options

Tokio provides different runtime configurations depending on your security tool’s needs. Choosing the right runtime impacts performance, memory usage, and behavior.

Multi-Threaded Runtime (Default)

// Default: Multi-threaded runtime
#[tokio::main]
async fn main() {
    // Uses work-stealing thread pool
    // Number of threads = CPU cores
}

// Explicit configuration:
#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
async fn main() {
    // 4 worker threads for async tasks
}

When to use:

• ✅ Most security tools (port scanners, web scrapers)
• ✅ Mixed I/O and CPU workloads
• ✅ Need parallelism across CPU cores
• ✅ Default choice for production

Characteristics:

• Work-stealing scheduler (tasks can move between threads)
• Automatic load balancing
• Higher memory overhead (multiple stacks)
• Best throughput for mixed workloads

Current-Thread Runtime (Single-Threaded)

#[tokio::main(flavor = "current_thread")]
async fn main() {
    // All tasks run on single thread
}

When to use:

• ✅ Simple scripts and tools
• ✅ Embedded systems (limited resources)
• ✅ Debugging (simpler to trace)
• ✅ No CPU-intensive work

Characteristics:

• Lower memory overhead
• Simpler debugging (no thread interleaving)
• No parallelism (tasks run sequentially)
• Good for pure I/O workloads

Manual Runtime Configuration

use tokio::runtime::{Builder, Runtime};

fn create_custom_runtime() -> Runtime {
    Builder::new_multi_thread()
        .worker_threads(8)              // Override CPU core count
        .thread_name("security-scanner") // Named threads for debugging
        .thread_stack_size(3 * 1024 * 1024) // 3 MB stack (default: platform-specific)
        .enable_all()                    // Enable I/O and time drivers
        .build()
        .unwrap()
}

fn main() {
    let runtime = create_custom_runtime();
    
    runtime.block_on(async {
        // Your async security tool
        run_scanner().await;
    });
}

Blocking Thread Pool Configuration

Critical for CPU-Intensive Work in Async Context

Tokio maintains a separate thread pool for spawn_blocking() operations:

use tokio::runtime::Builder;

fn main() {
    let runtime = Builder::new_multi_thread()
        .worker_threads(4)              // Async worker threads
        .max_blocking_threads(512)      // Blocking thread pool (default: 512)
        .build()
        .unwrap();
    
    runtime.block_on(async {
        // CPU-intensive work goes to blocking pool
        let hash = tokio::task::spawn_blocking(|| {
            bcrypt::hash("password", 12).unwrap()
        }).await.unwrap();
    });
}

Blocking Thread Pool Tuning:

Runtime Configuration Examples

Port Scanner (I/O-Heavy):

let runtime = Builder::new_multi_thread()
    .worker_threads(4)              // 4 async workers
    .thread_name("port-scanner")
    .max_blocking_threads(32)       // Rare blocking operations
    .enable_all()
    .build()
    .unwrap();

Log Analyzer (CPU-Heavy):

let runtime = Builder::new_multi_thread()
    .worker_threads(num_cpus::get()) // All CPU cores
    .thread_name("log-analyzer")
    .max_blocking_threads(512)       // Heavy spawn_blocking usage
    .thread_stack_size(4 * 1024 * 1024) // Larger stack for parsing
    .enable_all()
    .build()
    .unwrap();

Lightweight EDR Agent:

let runtime = Builder::new_multi_thread()
    .worker_threads(2)              // Minimal overhead
    .thread_name("edr-agent")
    .max_blocking_threads(64)       // Some system calls
    .enable_all()
    .build()
    .unwrap();

Performance Tuning Guidelines

Worker Thread Count:

• Default (CPU cores): Good starting point
• I/O-bound tools: CPU cores or slightly more (4-8)
• Mixed workload: CPU cores
• Low overhead agents: 2-4 threads

Blocking Thread Pool:

• Default (512): Works for most cases
• Rare blocking: 32-64 threads
• Heavy CPU work: 512-1024 threads
• Memory-constrained: 64-128 threads

Thread Stack Size:

• Default: Platform-specific (usually 2 MB)
• Deep recursion: 4-8 MB
• Memory-constrained: 1-2 MB
• Normal usage: Don’t change

Monitoring Runtime Health

use tokio::runtime::Handle;

async fn monitor_runtime_health() {
    let handle = Handle::current();
    
    // Get runtime metrics (requires `tokio_unstable` flag)
    loop {
        tokio::time::sleep(Duration::from_secs(10)).await;
        
        // Check for blocking operations
        // (Real metrics require tokio_unstable + metrics crate)
        println!("Runtime healthy");
    }
}

Common Runtime Pitfalls

❌ Mistake: Too many worker threads

// ❌ BAD: 64 threads for I/O-bound work (wasted overhead)
Builder::new_multi_thread()
    .worker_threads(64) // Overkill for async I/O!
    .build()

✅ Fix:

// ✅ GOOD: CPU core count is sufficient
Builder::new_multi_thread()
    .worker_threads(num_cpus::get()) // Optimal for most cases
    .build()

❌ Mistake: Blocking pool too small

// ❌ BAD: 8 blocking threads for heavy CPU work
Builder::new_multi_thread()
    .max_blocking_threads(8) // spawn_blocking will queue!
    .build()

✅ Fix:

// ✅ GOOD: Larger pool for CPU-intensive spawn_blocking
Builder::new_multi_thread()
    .max_blocking_threads(512) // Default, handles bursts
    .build()

Key Takeaways

1. Start with defaults (#[tokio::main]), tune if needed
2. Worker threads = CPU cores for most security tools
3. Blocking pool size matters if using spawn_blocking heavily
4. Current-thread runtime only for simple scripts
5. Monitor and profile before tuning (don’t guess!)

Golden Rule: Tokio’s defaults are excellent for 90% of security tools. Only tune if profiling shows a bottleneck.

Basic Async Function

use tokio::time::{sleep, Duration};

async fn process_security_event() {
    println!("Processing event...");
    sleep(Duration::from_secs(1)).await;
    println!("Event processed");
}

#[tokio::main]
async fn main() {
    process_security_event().await;
}

⚠️ CRITICAL: Blocking in Async Context (Most Common Mistake)

This is the #1 async Rust mistake in security tools.

When you call a blocking function inside an async context, you block the entire async runtime thread, preventing other tasks from running.

❌ What NOT to Do

use std::thread;
use std::time::Duration;

// ❌ BAD: Blocking sleep in async function
async fn bad_scan_port(port: u16) {
    println!("Scanning port {}", port);
    // This BLOCKS the async runtime thread for 1 second!
    thread::sleep(Duration::from_secs(1));
    println!("Port {} done", port);
}

// ❌ BAD: Blocking file I/O in async
async fn bad_read_config() -> String {
    // This BLOCKS the async runtime!
    std::fs::read_to_string("config.txt").unwrap()
}

// ❌ BAD: Blocking network call in async
async fn bad_http_request() -> String {
    // This BLOCKS the async runtime!
    reqwest::blocking::get("https://api.example.com")
        .unwrap()
        .text()
        .unwrap()
}

// ❌ BAD: CPU-intensive work in async
async fn bad_hash_password(password: &str) -> String {
    // This BLOCKS the runtime for seconds!
    bcrypt::hash(password, 12).unwrap()
}

// ❌ BAD: std::sync::Mutex in async (can deadlock)
use std::sync::Mutex;
async fn bad_shared_state(data: Arc<Mutex<Vec<String>>>) {
    let mut d = data.lock().unwrap(); // Can block indefinitely!
    d.push("value".to_string());
}

Impact of Blocking:

• 🚫 Runtime stalls: Other async tasks can’t run
• 🚫 Latency spikes: All tasks wait for the blocker
• 🚫 Throughput collapse: 10,000 → 10 concurrent tasks
• 🚫 Deadlocks: With std::sync primitives

Real-World Example:

// ❌ BAD: Port scanner that blocks
#[tokio::main]
async fn main() {
    let mut handles = vec![];
    
    for port in 1..=1000 {
        handles.push(tokio::spawn(async move {
            // BLOCKS runtime thread for 1 second!
            thread::sleep(Duration::from_secs(1));
            scan_port(port)
        }));
    }
    
    // Expected: 1000 scans in ~1 second (concurrent)
    // Actual: 1000 seconds (sequential, runtime blocked!)
}

✅ Correct Async Patterns

use tokio::time::{sleep, Duration};
use tokio::fs;
use tokio::sync::Mutex;

// ✅ GOOD: Async sleep
async fn good_scan_port(port: u16) {
    println!("Scanning port {}", port);
    // Yields to runtime, other tasks can run
    tokio::time::sleep(Duration::from_secs(1)).await;
    println!("Port {} done", port);
}

// ✅ GOOD: Async file I/O
async fn good_read_config() -> Result<String, std::io::Error> {
    // Non-blocking file read
    tokio::fs::read_to_string("config.txt").await
}

// ✅ GOOD: Async HTTP with async client
async fn good_http_request() -> Result<String, reqwest::Error> {
    // Non-blocking HTTP request
    reqwest::get("https://api.example.com")
        .await?
        .text()
        .await
}

// ✅ GOOD: Offload CPU work to blocking pool
async fn good_hash_password(password: String) -> Result<String, ()> {
    // Runs on separate blocking thread pool
    tokio::task::spawn_blocking(move || {
        bcrypt::hash(&password, 12).map_err(|_| ())
    })
    .await
    .unwrap()
}

// ✅ GOOD: tokio::sync::Mutex for async
async fn good_shared_state(data: Arc<Mutex<Vec<String>>>) {
    // Async-aware mutex, yields if locked
    let mut d = data.lock().await;
    d.push("value".to_string());
}

Using spawn_blocking for CPU-Intensive Work

use tokio::task;

// ✅ Pattern: Offload blocking work
async fn process_security_logs(logs: Vec<String>) -> Vec<Threat> {
    // Async: Receive logs from network
    let logs = fetch_logs().await;
    
    // CPU-intensive: Offload to blocking pool
    let threats = task::spawn_blocking(move || {
        logs.into_iter()
            .filter_map(|log| parse_threat(&log))
            .collect::<Vec<_>>()
    })
    .await
    .unwrap();
    
    // Async: Send alerts
    for threat in &threats {
        send_alert(threat).await;
    }
    
    threats
}

Common Blocking Operations to Watch For

Key Rules

1. Never use std::thread::sleep() in async → Use tokio::time::sleep().await
2. Never use std::fs in async → Use tokio::fs
3. Never use std::sync::Mutex in async → Use tokio::sync::Mutex
4. Never use reqwest::blocking in async → Use async reqwest
5. CPU-intensive work? → Use tokio::task::spawn_blocking()

Security Tool Impact:

A port scanner with blocking calls can degrade from 10,000 ports/sec to 10 ports/sec — a 1000x slowdown!

Spawning Tasks

#[tokio::main]
async fn main() {
    let handle = tokio::spawn(async {
        // Background task
        println!("Task running");
    });
    
    handle.await.unwrap();
}

Concurrent Processing

Parallel Processing

use futures::stream::{self, StreamExt};

async fn scan_ports_concurrent(ports: Vec<u16>) -> Vec<u16> {
    stream::iter(ports)
        .map(|port| async move {
            // Scan port
            check_port(port).await
        })
        .buffer_unordered(100)  // Process 100 concurrently
        .collect()
        .await
}

Shared State

use std::sync::Arc;
use tokio::sync::Mutex;

#[tokio::main]
async fn main() {
    let shared_state = Arc::new(Mutex::new(0));
    
    for i in 0..10 {
        let state = Arc::clone(&shared_state);
        tokio::spawn(async move {
            let mut value = state.lock().await;
            *value += i;
        });
    }
}

⚠️ Mutex vs RwLock in Async: Critical Choice

This impacts security tool performance significantly.

Mutex vs RwLock Decision Guide

Mutex (Exclusive Access)

use tokio::sync::Mutex;
use std::sync::Arc;

// ✅ GOOD: Frequent writes to shared counter
let counter = Arc::new(Mutex::new(0));

for _ in 0..1000 {
    let counter = counter.clone();
    tokio::spawn(async move {
        let mut c = counter.lock().await; // Exclusive lock
        *c += 1;
    });
}

When to use Mutex:

• ✅ Frequent writes (every access modifies state)
• ✅ Simpler logic (no read/write distinction)
• ✅ Low contention (locks held briefly)

RwLock (Multiple Readers, Exclusive Writer)

use tokio::sync::RwLock;
use std::sync::Arc;
use std::collections::HashMap;

// ✅ GOOD: Read-heavy threat intel cache
let threat_db = Arc::new(RwLock::new(HashMap::new()));

// Many readers (concurrent, no blocking)
for ip in ips {
    let db = threat_db.clone();
    tokio::spawn(async move {
        let db = db.read().await; // Shared lock (many readers)
        if db.contains_key(&ip) {
            println!("IP {} is malicious", ip);
        }
    });
}

// Single writer (exclusive, blocks readers)
{
    let mut db = threat_db.write().await; // Exclusive lock
    db.insert("1.2.3.4", ThreatInfo { severity: "high" });
}

When to use RwLock:

• ✅ Many reads, few writes (90%+ reads)
• ✅ Reads don’t modify state (lookups, queries)
• ✅ Parallel reads improve performance

Performance Comparison

Scenario: Threat Intel Lookup (1000 reads, 10 writes)

Common Mistakes

❌ Mistake 1: Using std::sync in async

use std::sync::Mutex; // ❌ WRONG! Blocks async runtime

let data = Arc::new(Mutex::new(vec![]));
tokio::spawn(async move {
    let mut d = data.lock().unwrap(); // Can deadlock!
    d.push(42);
});

✅ Fix:

use tokio::sync::Mutex; // ✅ CORRECT! Async-aware

let data = Arc::new(Mutex::new(vec![]));
tokio::spawn(async move {
    let mut d = data.lock().await; // Yields to runtime
    d.push(42);
});

❌ Mistake 2: RwLock for write-heavy workloads

// ❌ BAD: RwLock overhead for frequent writes
let counter = Arc::new(RwLock::new(0));
for _ in 0..10000 {
    let c = counter.clone();
    tokio::spawn(async move {
        let mut val = c.write().await; // Exclusive anyway!
        *val += 1;
    });
}

✅ Fix:

// ✅ GOOD: Mutex is simpler and faster for writes
let counter = Arc::new(Mutex::new(0));
for _ in 0..10000 {
    let c = counter.clone();
    tokio::spawn(async move {
        let mut val = c.lock().await;
        *val += 1;
    });
}

Lock-Free Alternative: DashMap

For high contention, consider lock-free data structures:

use dashmap::DashMap;
use std::sync::Arc;

// ✅ BEST: Lock-free concurrent HashMap
let threat_db = Arc::new(DashMap::new());

// Concurrent reads (no locks!)
for ip in ips {
    let db = threat_db.clone();
    tokio::spawn(async move {
        if let Some(info) = db.get(&ip) {
            println!("IP {} is malicious", ip);
        }
    });
}

// Concurrent writes (lock-free internally)
threat_db.insert("1.2.3.4", ThreatInfo { severity: "high" });

When to use DashMap:

• ✅ High contention (many concurrent accesses)
• ✅ HashMap-like data structure needed
• ✅ Performance critical path

Key Rules

1. Never use std::sync::Mutex in async → Use tokio::sync::Mutex
2. Never use std::sync::RwLock in async → Use tokio::sync::RwLock
3. Frequent writes? → Use Mutex
4. Many reads, few writes? → Use RwLock
5. High contention? → Consider DashMap or other lock-free structures

Async I/O Patterns

Network Operations

use tokio::net::TcpStream;

async fn connect_and_send(addr: &str) -> Result<(), Box<dyn std::error::Error>> {
    let mut stream = TcpStream::connect(addr).await?;
    stream.write_all(b"data").await?;
    Ok(())
}

File Operations

use tokio::fs::File;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

async fn process_file() -> Result<(), Box<dyn std::error::Error>> {
    let mut file = File::open("data.txt").await?;
    let mut contents = Vec::new();
    file.read_to_end(&mut contents).await?;
    Ok(())
}

Error Handling

Async Result Handling

async fn process_with_error() -> Result<String, Box<dyn std::error::Error>> {
    let result = risky_operation().await?;
    Ok(result)
}

async fn handle_errors() {
    match process_with_error().await {
        Ok(value) => println!("Success: {}", value),
        Err(e) => eprintln!("Error: {}", e),
    }
}

Advanced Patterns

Select Multiple Futures

use tokio::select;

async fn listen_multiple_sources() {
    let mut chan1 = receiver1();
    let mut chan2 = receiver2();
    
    loop {
        select! {
            msg = chan1.recv() => {
                handle_message1(msg).await;
            }
            msg = chan2.recv() => {
                handle_message2(msg).await;
            }
        }
    }
}

Task Cancellation and Graceful Shutdown

Critical for Production Security Tools

Security tools must handle shutdown gracefully to avoid:

• 🚫 Data loss (incomplete writes)
• 🚫 Resource leaks (open connections)
• 🚫 Corrupted state (partial updates)
• 🚫 Hanging processes (unkillable agents)

Basic Task Cancellation

use tokio::task::JoinHandle;

// Dropping a JoinHandle cancels the task
let handle: JoinHandle<_> = tokio::spawn(async {
    loop {
        println!("Running...");
        tokio::time::sleep(Duration::from_secs(1)).await;
    }
});

// Cancel by dropping
drop(handle); // Task is cancelled immediately

Graceful Shutdown with Signal Handling

use tokio::signal;
use tokio::sync::broadcast;

#[tokio::main]
async fn main() {
    let (shutdown_tx, _) = broadcast::channel(1);
    
    // Spawn security scanner tasks
    let mut handles = vec![];
    for i in 0..10 {
        let mut shutdown_rx = shutdown_tx.subscribe();
        
        handles.push(tokio::spawn(async move {
            loop {
                tokio::select! {
                    _ = shutdown_rx.recv() => {
                        println!("Task {} shutting down gracefully", i);
                        // Clean up resources
                        break;
                    }
                    _ = do_work() => {
                        // Normal work
                    }
                }
            }
        }));
    }
    
    // Wait for Ctrl+C
    signal::ctrl_c().await.unwrap();
    println!("Shutdown signal received");
    
    // Broadcast shutdown to all tasks
    let _ = shutdown_tx.send(());
    
    // Wait for all tasks to finish
    for handle in handles {
        handle.await.unwrap();
    }
    
    println!("All tasks shut down gracefully");
}

Production EDR Agent Shutdown Pattern

use tokio::sync::broadcast;
use tokio::signal;
use tokio::time::{timeout, Duration};

struct EdrAgent {
    shutdown_tx: broadcast::Sender<()>,
}

impl EdrAgent {
    fn new() -> Self {
        let (shutdown_tx, _) = broadcast::channel(1);
        Self { shutdown_tx }
    }
    
    async fn run(&self) {
        let mut shutdown_rx = self.shutdown_tx.subscribe();
        
        // Monitor events
        let monitor = tokio::spawn(async move {
            loop {
                tokio::select! {
                    _ = shutdown_rx.recv() => {
                        println!("Monitor shutting down");
                        break;
                    }
                    event = capture_event() => {
                        process_event(event).await;
                    }
                }
            }
        });
        
        // Wait for Ctrl+C
        signal::ctrl_c().await.unwrap();
        println!("Shutdown initiated");
        
        // Signal all tasks to stop
        let _ = self.shutdown_tx.send(());
        
        // Wait for tasks with timeout
        match timeout(Duration::from_secs(5), monitor).await {
            Ok(_) => println!("Clean shutdown"),
            Err(_) => println!("Force shutdown after timeout"),
        }
    }
}

Cancellation Safety

Some operations are NOT cancellation-safe:

// ❌ NOT cancellation-safe: recv() in select!
tokio::select! {
    msg = rx.recv() => {
        // If other branch completes first, msg is lost!
    }
    _ = timeout => {}
}

// ✅ Cancellation-safe: Use recv() only once
let msg = tokio::select! {
    msg = rx.recv() => msg,
    _ = timeout => None,
};

if let Some(msg) = msg {
    process(msg);
}

Shutdown Timeout Pattern

use tokio::time::{timeout, Duration};

async fn shutdown_with_timeout(handles: Vec<JoinHandle<()>>) {
    let shutdown_future = async {
        for handle in handles {
            handle.await.unwrap();
        }
    };
    
    match timeout(Duration::from_secs(10), shutdown_future).await {
        Ok(_) => println!("Graceful shutdown completed"),
        Err(_) => {
            println!("Shutdown timeout - forcing exit");
            std::process::exit(1);
        }
    }
}

Logging Overhead in Async Hot Paths

Critical Performance Consideration

Logging can destroy async performance if not handled carefully.

❌ Bad: Blocking Logging in Async

use log::info;

async fn scan_port(port: u16) -> bool {
    // ❌ BAD: Synchronous logging blocks async runtime
    info!("Scanning port {}", port);
    
    let result = TcpStream::connect(("127.0.0.1", port)).await.is_ok();
    
    // ❌ BAD: More blocking I/O
    info!("Port {} result: {}", port, result);
    
    result
}

Impact:

• Logging to file/stdout is synchronous I/O
• Blocks async runtime thread
• 10,000 log lines = seconds of blocking

✅ Good: Async-Aware Logging with tracing

use tracing::{info, instrument};

#[instrument(skip(port))] // Minimal overhead
async fn scan_port(port: u16) -> bool {
    // ✅ GOOD: tracing is async-aware
    info!("Scanning port {}", port);
    
    let result = TcpStream::connect(("127.0.0.1", port)).await.is_ok();
    
    result
}

// Setup tracing with async-friendly subscriber
use tracing_subscriber;

fn main() {
    tracing_subscriber::fmt()
        .with_max_level(tracing::Level::INFO)
        .init();
}

Hot Path Optimization: Conditional Logging

// ✅ BEST: Skip logging in hot path, only log errors
async fn scan_port_optimized(port: u16) -> bool {
    let result = TcpStream::connect(("127.0.0.1", port)).await;
    
    // Only log failures (rare)
    if let Err(e) = &result {
        tracing::warn!("Port {} failed: {}", port, e);
    }
    
    result.is_ok()
}

Logging Overhead Comparison

Production Logging Pattern

use tracing::{info, warn, error, debug};

// ✅ Production-ready pattern
async fn security_scan(target: &str) {
    // Info: Important milestones only
    info!("Starting scan of {}", target);
    
    for port in 1..=65535 {
        // Debug: Hot path (disabled in release by default)
        debug!("Checking port {}", port);
        
        match scan_port(port).await {
            Ok(true) => {
                // Warn: Interesting findings
                warn!("Open port found: {}", port);
            }
            Err(e) => {
                // Error: Actual errors
                error!("Scan error on port {}: {}", port, e);
            }
            _ => {} // Silent for closed ports (most cases)
        }
    }
    
    info!("Scan completed");
}

Key Takeaways

Graceful Shutdown:

1. Use broadcast::channel for shutdown signals
2. Handle Ctrl+C with tokio::signal
3. Set timeouts for shutdown (force exit if needed)
4. Clean up resources before exiting

Logging in Async:

1. Use tracing crate (async-aware)
2. Avoid logging in hot paths (10k+ calls/sec)
3. Log failures only, not successes
4. Use debug logging for hot paths (disabled in release)

Troubleshooting Guide

Problem: Tasks Not Completing

Solution:

• Check for blocking operations
• Verify .await is used
• Check for deadlocks
• Review task spawning

Problem: High Memory Usage

Solution:

• Limit concurrent tasks
• Use backpressure
• Monitor task count
• Optimize buffer sizes

Backpressure and Bounded Concurrency

Critical for Production Security Tools

Without backpressure, async security tools can:

• 🚫 Exhaust memory (OOM kills)
• 🚫 Overwhelm targets (accidental DoS)
• 🚫 Trigger IDS alerts (too aggressive)
• 🚫 Crash EDR agents (unbounded queues)

The Problem: Unbounded Concurrency

// ❌ BAD: No backpressure - spawns 1 million tasks!
#[tokio::main]
async fn main() {
    let mut handles = vec![];
    
    // Spawns ALL tasks immediately - memory explosion!
    for i in 0..1_000_000 {
        handles.push(tokio::spawn(async move {
            scan_target(i).await;
        }));
    }
    
    // Waits for all (but memory already exhausted)
    for handle in handles {
        handle.await.unwrap();
    }
}

Impact:

• 1 million tasks × 4 KB stack = 4 GB memory
• System OOMs, kills your security agent
• No rate limiting, floods network

Solution 1: Bounded Concurrency with buffer_unordered

use futures::stream::{self, StreamExt};

// ✅ GOOD: Limited concurrency
async fn scan_with_backpressure(targets: Vec<String>) -> Vec<ScanResult> {
    stream::iter(targets)
        .map(|target| async move {
            scan_target(&target).await
        })
        .buffer_unordered(100)  // Only 100 concurrent tasks!
        .collect()
        .await
}

How it works:

• Creates iterator of 1 million targets
• But only 100 are in-flight at once
• As tasks complete, new ones start
• Constant memory usage

Solution 2: Semaphore-Based Rate Limiting

use tokio::sync::Semaphore;
use std::sync::Arc;

// ✅ GOOD: Semaphore controls concurrency
async fn scan_with_semaphore(targets: Vec<String>) -> Vec<ScanResult> {
    let semaphore = Arc::new(Semaphore::new(100)); // Max 100 concurrent
    let mut handles = vec![];
    
    for target in targets {
        let permit = semaphore.clone().acquire_owned().await.unwrap();
        
        handles.push(tokio::spawn(async move {
            let result = scan_target(&target).await;
            drop(permit); // Release permit when done
            result
        }));
    }
    
    // Collect results
    let mut results = vec![];
    for handle in handles {
        results.push(handle.await.unwrap());
    }
    results
}

Solution 3: Channel-Based Backpressure

use tokio::sync::mpsc;

// ✅ GOOD: Bounded channel provides backpressure
async fn scan_with_channels(targets: Vec<String>) -> Vec<ScanResult> {
    let (tx, mut rx) = mpsc::channel(100); // Buffer only 100 tasks
    
    // Producer task (sends targets)
    tokio::spawn(async move {
        for target in targets {
            // Blocks if channel is full (backpressure!)
            tx.send(target).await.unwrap();
        }
    });
    
    // Consumer tasks (process targets)
    let mut handles = vec![];
    for _ in 0..100 {
        let mut rx = rx.clone();
        handles.push(tokio::spawn(async move {
            let mut results = vec![];
            while let Some(target) = rx.recv().await {
                results.push(scan_target(&target).await);
            }
            results
        }));
    }
    
    // Collect results from all workers
    let mut all_results = vec![];
    for handle in handles {
        all_results.extend(handle.await.unwrap());
    }
    all_results
}

Real-World Example: Port Scanner with Backpressure

use futures::stream::{self, StreamExt};
use std::net::{IpAddr, SocketAddr};
use tokio::net::TcpStream;
use tokio::time::{timeout, Duration};

#[derive(Debug)]
struct ScanResult {
    addr: SocketAddr,
    open: bool,
}

async fn scan_port(addr: SocketAddr) -> ScanResult {
    let open = timeout(
        Duration::from_secs(1),
        TcpStream::connect(addr)
    ).await.is_ok();
    
    ScanResult { addr, open }
}

async fn scan_network(ip: IpAddr, ports: Vec<u16>) -> Vec<ScanResult> {
    let targets: Vec<SocketAddr> = ports
        .iter()
        .map(|&port| SocketAddr::new(ip, port))
        .collect();
    
    // ✅ Backpressure: Only 500 concurrent connections
    stream::iter(targets)
        .map(|addr| async move {
            scan_port(addr).await
        })
        .buffer_unordered(500)  // Tune based on system limits
        .collect()
        .await
}

#[tokio::main]
async fn main() {
    let ip = "192.168.1.1".parse().unwrap();
    let ports: Vec<u16> = (1..=65535).collect();
    
    let results = scan_network(ip, ports).await;
    
    let open_ports: Vec<_> = results
        .into_iter()
        .filter(|r| r.open)
        .collect();
    
    println!("Open ports: {:?}", open_ports);
}

Choosing Concurrency Limits

Backpressure Comparison

Memory Usage Comparison

Unbounded:
Tasks: [||||||||||||||||||||||||||||||||] 1,000,000 tasks
Memory: 4 GB+ → OOM crash

With Backpressure (buffer_unordered(100)):
Tasks: [||||] 100 tasks (constant)
Memory: 400 KB → Stable

Key Takeaways

1. Always use bounded concurrency in production security tools
2. Start conservative (100-500), tune based on profiling
3. Monitor memory usage to detect unbounded growth
4. Backpressure prevents:
   • Memory exhaustion (OOM kills)
   • Network flooding (accidental DoS)
   • Detection by IDS (rate anomalies)
   • System instability (resource exhaustion)

Security Tool Golden Rule:

“If your async security tool doesn’t have backpressure, it will crash in production.” — Every security engineer who learned the hard way

Advanced Scenarios

Scenario 1: Basic Async Rust Security Tool

Objective: Build basic async Rust security tool. Steps: Set up async runtime, implement async functions, test functionality. Expected: Basic async tool operational.

Scenario 2: Intermediate Advanced Async Patterns

Objective: Implement advanced async patterns. Steps: Task spawning + channels + backpressure + error handling. Expected: Advanced async patterns operational.

Scenario 3: Advanced Comprehensive Async Security Tool

Objective: Complete async security tool program. Steps: All async features + performance optimization + testing + maintenance. Expected: Comprehensive async tool.

Theory and “Why” Async Rust Works for Security

Why Async Improves Performance

• Non-blocking I/O
• Concurrent task execution
• Better resource utilization
• Scalable architecture

Why Rust’s Async is Safe

• Compile-time safety guarantees
• No data races
• Memory safety
• Type safety

Comprehensive Troubleshooting

Issue: Deadlocks in Async Code

Diagnosis: Review task spawning, check locks, analyze blocking operations. Solutions: Avoid blocking in async, use async locks, review task design.

Issue: High Memory Usage

Diagnosis: Monitor task count, check buffers, analyze memory usage. Solutions: Limit concurrent tasks, use backpressure, optimize buffers.

Issue: Performance Issues

Diagnosis: Profile async code, check task overhead, measure performance. Solutions: Optimize tasks, reduce overhead, improve efficiency.

Code Review Checklist for Async Rust Security

Async Patterns

• Proper use of .await (no blocking in async code)
• Task spawning with appropriate concurrency limits
• Channel usage for safe communication
• Backpressure handling for resource management

Error Handling

• Async error handling with Result types
• Proper error propagation in async chains
• Timeout handling for async operations
• Graceful shutdown of async tasks

Resource Management

• Proper cleanup of async resources
• Task cancellation when needed
• Memory-conscious async code (avoid unbounded buffers)
• Connection pooling where appropriate

Safety

• No data races in concurrent async code
• Proper synchronization with Arc/Mutex if needed
• Avoid blocking operations in async context
• Test async code thoroughly

Performance

• Appropriate concurrency levels
• Efficient async I/O operations
• Proper use of async runtime features
• Performance profiling and optimization

Complete Code Examples

Example 1: Async Web Vulnerability Scanner

[package]
name = "async-web-scanner"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1.40", features = ["full"] }
reqwest = "0.11"
futures = "0.3"
clap = { version = "4.5", features = ["derive"] }
tracing = "0.1"
tracing-subscriber = "0.3"
anyhow = "1.0"

use anyhow::Result;
use clap::Parser;
use futures::stream::{self, StreamExt};
use reqwest::Client;
use std::time::Duration;
use tracing::{info, warn};

#[derive(Parser)]
struct Args {
    /// Target URL
    #[arg(short, long)]
    url: String,

    /// Concurrent requests
    #[arg(short, long, default_value_t = 50)]
    concurrency: usize,
}

#[derive(Debug)]
struct VulnCheck {
    path: String,
    status: u16,
    vulnerable: bool,
}

async fn check_path(client: &Client, base_url: &str, path: &str) -> Result<VulnCheck> {
    let url = format!("{}{}", base_url, path);
    
    let response = client
        .get(&url)
        .timeout(Duration::from_secs(5))
        .send()
        .await?;
    
    let status = response.status().as_u16();
    let vulnerable = status == 200;
    
    if vulnerable {
        warn!("⚠️  Potential vulnerability: {} ({})", path, status);
    }
    
    Ok(VulnCheck {
        path: path.to_string(),
        status,
        vulnerable,
    })
}

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt::init();
    let args = Args::parse();
    
    let client = Client::builder()
        .timeout(Duration::from_secs(10))
        .build()?;
    
    // Common vulnerable paths
    let paths = vec![
        "/.git/config",
        "/.env",
        "/admin",
        "/phpmyadmin",
        "/.aws/credentials",
        "/backup.sql",
        "/config.php.bak",
        "/web.config",
    ];
    
    info!("Scanning {} with {} concurrent requests", args.url, args.concurrency);
    
    let results: Vec<VulnCheck> = stream::iter(paths)
        .map(|path| {
            let client = client.clone();
            let url = args.url.clone();
            async move {
                check_path(&client, &url, path).await.ok()
            }
        })
        .buffer_unordered(args.concurrency)
        .filter_map(|r| async { r })
        .collect()
        .await;
    
    let vulnerable: Vec<_> = results.iter().filter(|r| r.vulnerable).collect();
    
    println!("\n{}", "=".repeat(60));
    println!("SCAN RESULTS");
    println!("{}", "=".repeat(60));
    println!("Total paths checked: {}", results.len());
    println!("Vulnerable paths: {}", vulnerable.len());
    
    if !vulnerable.is_empty() {
        println!("\nVULNERABLE PATHS:");
        for vuln in vulnerable {
            println!("  {} (HTTP {})", vuln.path, vuln.status);
        }
    }
    
    Ok(())
}

Example 2: Async Log Analyzer with Hybrid Approach

[package]
name = "async-log-analyzer"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1.40", features = ["full"] }
rayon = "1.8"
regex = "1.10"
clap = { version = "4.5", features = ["derive"] }
tracing = "0.1"
tracing-subscriber = "0.3"
anyhow = "1.0"

use anyhow::Result;
use clap::Parser;
use rayon::prelude::*;
use regex::Regex;
use std::sync::Arc;
use tokio::fs::File;
use tokio::io::{AsyncBufReadExt, BufReader};
use tracing::{info, warn};

#[derive(Parser)]
struct Args {
    /// Log file path
    #[arg(short, long)]
    file: String,
}

#[derive(Debug, Clone)]
struct ThreatEvent {
    line_number: usize,
    ip: String,
    event_type: String,
    severity: String,
}

fn analyze_line(line_num: usize, line: &str) -> Option<ThreatEvent> {
    // Detect SQL injection attempts
    let sql_injection = Regex::new(r"(?i)(union|select|insert|drop|delete)\s+(from|into|table)").unwrap();
    
    // Detect XSS attempts
    let xss = Regex::new(r"(?i)<script|javascript:|onerror=").unwrap();
    
    // Extract IP (simple pattern)
    let ip_regex = Regex::new(r"\b\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}\b").unwrap();
    let ip = ip_regex.find(line)
        .map(|m| m.as_str().to_string())
        .unwrap_or_else(|| "unknown".to_string());
    
    if sql_injection.is_match(line) {
        Some(ThreatEvent {
            line_number: line_num,
            ip,
            event_type: "SQL Injection".to_string(),
            severity: "HIGH".to_string(),
        })
    } else if xss.is_match(line) {
        Some(ThreatEvent {
            line_number: line_num,
            ip,
            event_type: "XSS Attempt".to_string(),
            severity: "MEDIUM".to_string(),
        })
    } else {
        None
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt::init();
    let args = Args::parse();
    
    info!("Reading log file: {}", args.file);
    
    // ✅ HYBRID APPROACH:
    // 1. Async: Read file line by line (I/O-bound)
    let file = File::open(&args.file).await?;
    let reader = BufReader::new(file);
    let mut lines_stream = reader.lines();
    
    let mut all_lines = Vec::new();
    let mut line_num = 0;
    
    while let Some(line) = lines_stream.next_line().await? {
        line_num += 1;
        all_lines.push((line_num, line));
    }
    
    info!("Read {} lines, analyzing...", all_lines.len());
    
    // 2. Rayon: Analyze lines in parallel (CPU-bound)
    let threats: Vec<ThreatEvent> = tokio::task::spawn_blocking(move || {
        all_lines
            .par_iter()
            .filter_map(|(num, line)| analyze_line(*num, line))
            .collect()
    })
    .await?;
    
    // 3. Async: Send alerts (I/O-bound)
    if !threats.is_empty() {
        warn!("Found {} threats!", threats.len());
        
        println!("\n{}", "=".repeat(80));
        println!("THREAT ANALYSIS REPORT");
        println!("{}", "=".repeat(80));
        println!("{:<10} {:<20} {:<20} {:<10}", "LINE", "IP", "THREAT", "SEVERITY");
        println!("{}", "-".repeat(80));
        
        for threat in &threats {
            println!(
                "{:<10} {:<20} {:<20} {:<10}",
                threat.line_number, threat.ip, threat.event_type, threat.severity
            );
        }
        
        println!("{}", "=".repeat(80));
    } else {
        info!("No threats detected");
    }
    
    Ok(())
}

Example 3: Async Network Traffic Monitor with Backpressure

[package]
name = "async-traffic-monitor"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1.40", features = ["full"] }
tokio-util = { version = "0.7", features = ["codec"] }
futures = "0.3"
clap = { version = "4.5", features = ["derive"] }
tracing = "0.1"
tracing-subscriber = "0.3"
anyhow = "1.0"
dashmap = "5.5"

use anyhow::Result;
use clap::Parser;
use dashmap::DashMap;
use std::net::SocketAddr;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::broadcast;
use tokio::time::interval;
use tracing::{info, warn, error};

#[derive(Parser)]
struct Args {
    /// Listen address
    #[arg(short, long, default_value = "127.0.0.1:8080")]
    listen: String,

    /// Max concurrent connections
    #[arg(short, long, default_value_t = 1000)]
    max_connections: usize,
}

#[derive(Debug, Clone)]
struct ConnectionStats {
    bytes_received: u64,
    bytes_sent: u64,
    requests: u64,
}

struct TrafficMonitor {
    stats: Arc<DashMap<SocketAddr, ConnectionStats>>,
    shutdown_tx: broadcast::Sender<()>,
}

impl TrafficMonitor {
    fn new() -> Self {
        let (shutdown_tx, _) = broadcast::channel(1);
        
        Self {
            stats: Arc::new(DashMap::new()),
            shutdown_tx,
        }
    }
    
    async fn handle_connection(
        &self,
        mut stream: TcpStream,
        addr: SocketAddr,
    ) -> Result<()> {
        let mut buffer = vec![0u8; 1024];
        
        loop {
            match stream.read(&mut buffer).await {
                Ok(0) => break, // Connection closed
                Ok(n) => {
                    // Update stats
                    self.stats
                        .entry(addr)
                        .and_modify(|s| {
                            s.bytes_received += n as u64;
                            s.requests += 1;
                        })
                        .or_insert(ConnectionStats {
                            bytes_received: n as u64,
                            bytes_sent: 0,
                            requests: 1,
                        });
                    
                    // Echo response
                    let response = b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nOK";
                    stream.write_all(response).await?;
                    
                    self.stats
                        .entry(addr)
                        .and_modify(|s| s.bytes_sent += response.len() as u64);
                }
                Err(e) => {
                    error!("Connection error from {}: {}", addr, e);
                    break;
                }
            }
        }
        
        Ok(())
    }
    
    async fn run(&self, listen_addr: &str, max_connections: usize) -> Result<()> {
        let listener = TcpListener::bind(listen_addr).await?;
        info!("Listening on {} (max {} connections)", listen_addr, max_connections);
        
        // ✅ Backpressure: Use semaphore to limit connections
        let semaphore = Arc::new(tokio::sync::Semaphore::new(max_connections));
        let mut shutdown_rx = self.shutdown_tx.subscribe();
        
        loop {
            tokio::select! {
                _ = shutdown_rx.recv() => {
                    info!("Shutdown signal received");
                    break;
                }
                result = listener.accept() => {
                    match result {
                        Ok((stream, addr)) => {
                            // Acquire permit (blocks if at limit)
                            let permit = semaphore.clone().acquire_owned().await.unwrap();
                            let monitor = self.clone();
                            
                            tokio::spawn(async move {
                                info!("New connection from {}", addr);
                                
                                if let Err(e) = monitor.handle_connection(stream, addr).await {
                                    error!("Error handling {}: {}", addr, e);
                                }
                                
                                drop(permit); // Release permit
                                info!("Connection closed: {}", addr);
                            });
                        }
                        Err(e) => {
                            error!("Accept error: {}", e);
                        }
                    }
                }
            }
        }
        
        Ok(())
    }
    
    async fn print_stats_loop(&self) {
        let mut ticker = interval(Duration::from_secs(10));
        
        loop {
            ticker.tick().await;
            
            println!("\n{}", "=".repeat(80));
            println!("TRAFFIC STATISTICS");
            println!("{}", "=".repeat(80));
            println!("{:<20} {:<15} {:<15} {:<10}", "ADDRESS", "BYTES RX", "BYTES TX", "REQUESTS");
            println!("{}", "-".repeat(80));
            
            let mut total_rx = 0u64;
            let mut total_tx = 0u64;
            let mut total_req = 0u64;
            
            for entry in self.stats.iter() {
                let addr = entry.key();
                let stats = entry.value();
                
                println!(
                    "{:<20} {:<15} {:<15} {:<10}",
                    addr, stats.bytes_received, stats.bytes_sent, stats.requests
                );
                
                total_rx += stats.bytes_received;
                total_tx += stats.bytes_sent;
                total_req += stats.requests;
            }
            
            println!("{}", "-".repeat(80));
            println!(
                "{:<20} {:<15} {:<15} {:<10}",
                "TOTAL", total_rx, total_tx, total_req
            );
            println!("{}", "=".repeat(80));
        }
    }
    
    fn shutdown_sender(&self) -> broadcast::Sender<()> {
        self.shutdown_tx.clone()
    }
}

impl Clone for TrafficMonitor {
    fn clone(&self) -> Self {
        Self {
            stats: self.stats.clone(),
            shutdown_tx: self.shutdown_tx.clone(),
        }
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt::init();
    let args = Args::parse();
    
    let monitor = TrafficMonitor::new();
    
    // Setup graceful shutdown
    let shutdown_tx = monitor.shutdown_sender();
    tokio::spawn(async move {
        tokio::signal::ctrl_c().await.unwrap();
        warn!("Shutdown signal received");
        let _ = shutdown_tx.send(());
    });
    
    // Spawn stats reporter
    let stats_monitor = monitor.clone();
    tokio::spawn(async move {
        stats_monitor.print_stats_loop().await;
    });
    
    // Run server
    monitor.run(&args.listen, args.max_connections).await?;
    
    Ok(())
}

# Terminal 1: Start monitor
cargo run -- --listen 127.0.0.1:8080 --max-connections 100

# Terminal 2: Send requests
for i in {1..1000}; do curl http://127.0.0.1:8080 & done

Cleanup

# Clean up all example projects
cd async-port-scanner && cargo clean
cd ../async-web-scanner && cargo clean
cd ../async-log-analyzer && cargo clean
cd ../async-traffic-monitor && cargo clean

# Remove build artifacts
find . -name "target" -type d -exec rm -rf {} +

Real-World Case Study

Complete Production-Ready Async Port Scanner

Full Implementation with All Best Practices

This is a complete, production-ready async port scanner demonstrating:

• Tokio runtime configuration
• Backpressure with bounded concurrency
• Graceful shutdown
• Async-aware logging
• Error handling
• Progress reporting

[package]
name = "async-port-scanner"
version = "0.1.0"
edition = "2021"

[dependencies]
tokio = { version = "1.40", features = ["full"] }
futures = "0.3"
clap = { version = "4.5", features = ["derive"] }
tracing = "0.1"
tracing-subscriber = { version = "0.3", features = ["env-filter"] }
anyhow = "1.0"
indicatif = "0.17"

use anyhow::Result;
use clap::Parser;
use futures::stream::{self, StreamExt};
use indicatif::{ProgressBar, ProgressStyle};
use std::net::{IpAddr, SocketAddr};
use std::sync::Arc;
use std::time::Duration;
use tokio::net::TcpStream;
use tokio::sync::broadcast;
use tokio::time::timeout;
use tracing::{info, warn, error, debug};

/// Async Port Scanner - Production-ready security tool
#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None)]
struct Args {
    /// Target IP address to scan
    #[arg(short, long)]
    target: String,

    /// Start port (default: 1)
    #[arg(short, long, default_value_t = 1)]
    start: u16,

    /// End port (default: 65535)
    #[arg(short, long, default_value_t = 65535)]
    end: u16,

    /// Concurrent connections (default: 500)
    #[arg(short, long, default_value_t = 500)]
    concurrency: usize,

    /// Connection timeout in milliseconds (default: 1000)
    #[arg(long, default_value_t = 1000)]
    timeout: u64,

    /// Verbose output
    #[arg(short, long)]
    verbose: bool,
}

#[derive(Debug, Clone)]
struct ScanResult {
    addr: SocketAddr,
    open: bool,
    duration_ms: u64,
}

struct PortScanner {
    target: IpAddr,
    start_port: u16,
    end_port: u16,
    concurrency: usize,
    timeout: Duration,
    shutdown_tx: broadcast::Sender<()>,
}

impl PortScanner {
    fn new(
        target: IpAddr,
        start_port: u16,
        end_port: u16,
        concurrency: usize,
        timeout_ms: u64,
    ) -> Self {
        let (shutdown_tx, _) = broadcast::channel(1);
        
        Self {
            target,
            start_port,
            end_port,
            concurrency,
            timeout: Duration::from_millis(timeout_ms),
            shutdown_tx,
        }
    }

    /// Scan a single port
    async fn scan_port(&self, port: u16) -> ScanResult {
        let addr = SocketAddr::new(self.target, port);
        let start = std::time::Instant::now();

        debug!("Scanning port {}", port);

        let open = timeout(self.timeout, TcpStream::connect(addr))
            .await
            .is_ok();

        let duration_ms = start.elapsed().as_millis() as u64;

        if open {
            info!("✓ Port {} is OPEN ({}ms)", port, duration_ms);
        }

        ScanResult {
            addr,
            open,
            duration_ms,
        }
    }

    /// Scan all ports with backpressure
    async fn scan_all(&self) -> Result<Vec<ScanResult>> {
        let total_ports = (self.end_port - self.start_port + 1) as u64;
        
        info!(
            "Starting scan of {} ports on {} (concurrency: {})",
            total_ports, self.target, self.concurrency
        );

        // Progress bar
        let progress = ProgressBar::new(total_ports);
        progress.set_style(
            ProgressStyle::default_bar()
                .template("[{elapsed_precise}] {bar:40.cyan/blue} {pos}/{len} ports ({per_sec}) ETA: {eta}")
                .unwrap()
                .progress_chars("=>-"),
        );

        // Create port range
        let ports: Vec<u16> = (self.start_port..=self.end_port).collect();

        // Scan with bounded concurrency (backpressure!)
        let results: Vec<ScanResult> = stream::iter(ports)
            .map(|port| {
                let scanner = self.clone();
                async move {
                    let result = scanner.scan_port(port).await;
                    result
                }
            })
            .buffer_unordered(self.concurrency) // ✅ Backpressure: only N concurrent
            .inspect(|_| {
                progress.inc(1);
            })
            .collect()
            .await;

        progress.finish_with_message("Scan complete");

        Ok(results)
    }

    /// Get shutdown signal sender
    fn shutdown_sender(&self) -> broadcast::Sender<()> {
        self.shutdown_tx.clone()
    }
}

// Implement Clone for PortScanner
impl Clone for PortScanner {
    fn clone(&self) -> Self {
        Self {
            target: self.target,
            start_port: self.start_port,
            end_port: self.end_port,
            concurrency: self.concurrency,
            timeout: self.timeout,
            shutdown_tx: self.shutdown_tx.clone(),
        }
    }
}

/// Print scan summary
fn print_summary(results: &[ScanResult]) {
    let open_ports: Vec<_> = results
        .iter()
        .filter(|r| r.open)
        .collect();

    println!("\n{}", "=".repeat(60));
    println!("SCAN SUMMARY");
    println!("{}", "=".repeat(60));
    println!("Total ports scanned: {}", results.len());
    println!("Open ports found: {}", open_ports.len());
    println!("Closed/filtered: {}", results.len() - open_ports.len());

    if !open_ports.is_empty() {
        println!("\nOPEN PORTS:");
        println!("{:<10} {:<20} {:<10}", "PORT", "ADDRESS", "RESPONSE");
        println!("{}", "-".repeat(60));
        
        for result in open_ports {
            println!(
                "{:<10} {:<20} {}ms",
                result.addr.port(),
                result.addr,
                result.duration_ms
            );
        }
    }

    // Calculate statistics
    let avg_duration: f64 = results
        .iter()
        .map(|r| r.duration_ms as f64)
        .sum::<f64>()
        / results.len() as f64;

    println!("\nSTATISTICS:");
    println!("Average response time: {:.2}ms", avg_duration);
    println!("{}", "=".repeat(60));
}

#[tokio::main]
async fn main() -> Result<()> {
    let args = Args::parse();

    // Setup logging
    let log_level = if args.verbose { "debug" } else { "info" };
    tracing_subscriber::fmt()
        .with_env_filter(log_level)
        .with_target(false)
        .with_thread_ids(false)
        .with_file(false)
        .with_line_number(false)
        .init();

    // Parse target IP
    let target: IpAddr = args.target.parse()
        .map_err(|_| anyhow::anyhow!("Invalid IP address: {}", args.target))?;

    // Validate port range
    if args.start > args.end {
        return Err(anyhow::anyhow!("Start port must be <= end port"));
    }

    // Create scanner
    let scanner = PortScanner::new(
        target,
        args.start,
        args.end,
        args.concurrency,
        args.timeout,
    );

    // Setup graceful shutdown
    let shutdown_tx = scanner.shutdown_sender();
    tokio::spawn(async move {
        tokio::signal::ctrl_c().await.unwrap();
        warn!("Shutdown signal received, stopping scan...");
        let _ = shutdown_tx.send(());
    });

    // Run scan
    let start_time = std::time::Instant::now();
    
    match scanner.scan_all().await {
        Ok(results) => {
            let elapsed = start_time.elapsed();
            
            print_summary(&results);
            
            info!(
                "Scan completed in {:.2}s ({:.0} ports/sec)",
                elapsed.as_secs_f64(),
                results.len() as f64 / elapsed.as_secs_f64()
            );
        }
        Err(e) => {
            error!("Scan failed: {}", e);
            return Err(e);
        }
    }

    Ok(())
}