sfo-pool

sfo-pool is a Tokio-based asynchronous worker pool that provides both general-purpose and keyed worker pools.

It manages workers with RAII: get_worker() returns a guard, and the worker is automatically returned to the pool when that guard goes out of scope. No explicit release call is required.

Features

Asynchronous worker creation with configurable pool capacity
Asynchronous queueing when no worker is available, with automatic wake-up when a worker is returned
Invalid-worker detection and eviction through Worker::is_work()
Idle timeouts and explicit idle-worker cleanup
Graceful pool clearing that waits for in-flight workers
Keyed worker reuse, creation, replacement, and limits
Automatic capacity-reservation rollback when creation fails or a creation task is cancelled

Installation

Add the following dependencies to Cargo.toml:

[dependencies]
sfo-pool = "0.3"
async-trait = "0.1"
tokio = { version = "1", features = ["rt-multi-thread", "macros"] }

General-purpose worker pool

Implement Worker and WorkerFactory, then acquire workers from a WorkerPool:

use async_trait::async_trait;
use sfo_pool::{PoolResult, Worker, WorkerFactory, WorkerPool};

struct Connection {
    id: u64,
    healthy: bool,
}

impl Worker for Connection {
    fn is_work(&self) -> bool {
        self.healthy
    }
}

struct ConnectionFactory;

#[async_trait]
impl WorkerFactory<Connection> for ConnectionFactory {
    async fn create(&self) -> PoolResult<Connection> {
        // A real connection can be established asynchronously here.
        Ok(Connection {
            id: 1,
            healthy: true,
        })
    }
}

#[tokio::main]
async fn main() -> PoolResult<()> {
    let pool = WorkerPool::new(8, ConnectionFactory);

    {
        let connection = pool.get_worker().await?;
        println!("using connection {}", connection.id);
    } // Dropping the guard automatically returns the connection to the pool.

    pool.clear_all_worker().await;
    Ok(())
}

Use new_with_config to configure an idle timeout:

use sfo_pool::{WorkerPool, WorkerPoolConfig};
use std::time::Duration;

# use async_trait::async_trait;
# use sfo_pool::{PoolResult, Worker, WorkerFactory};
# struct Connection;
# impl Worker for Connection { fn is_work(&self) -> bool { true } }
# struct ConnectionFactory;
# #[async_trait]
# impl WorkerFactory<Connection> for ConnectionFactory {
#     async fn create(&self) -> PoolResult<Connection> { Ok(Connection) }
# }
let pool = WorkerPool::new_with_config(
    ConnectionFactory,
    WorkerPoolConfig::default()
        .with_max_count(Some(8))
        .with_max_idle_count(Some(4))
        .with_idle_timeout(Some(Duration::from_secs(60))),
);

// Idle workers are cleaned up lazily before the next acquisition, or explicitly.
let removed_count = pool.cleanup_idle_worker();
# let _ = removed_count;

max_count limits the total of active, idle, and currently creating workers. Set it to None for an unlimited total; Some(0) is invalid. max_idle_count independently limits only the idle LRU cache. None preserves the previous behavior, Some(0) disables idle caching, and a full idle cache never prevents creating another active worker when total capacity permits. Configuration fields are private; construct values with Default and the with_* methods.

Keyed worker pool

The keyed pool is useful when workers must be allocated by tenant, region, database, or another condition:

use async_trait::async_trait;
use sfo_pool::{
    KeyedWorker, KeyedWorkerFactory, KeyedWorkerPool,
    KeyedWorkerPoolConfig, PoolResult,
};

#[derive(Clone, Debug, Hash, Eq, PartialEq)]
enum Region {
    East,
    West,
}

struct RegionalConnection {
    region: Region,
    healthy: bool,
}

impl KeyedWorker<Region> for RegionalConnection {
    fn is_work(&self) -> bool {
        self.healthy
    }

    fn supports(&self, region: Region) -> bool {
        self.region == region
    }

    fn primary_key(&self) -> Region {
        self.region.clone()
    }
}

struct RegionalConnectionFactory;

#[async_trait]
impl KeyedWorkerFactory<Region, RegionalConnection> for RegionalConnectionFactory {
    async fn create(&self, region: Region) -> PoolResult<RegionalConnection> {
        Ok(RegionalConnection {
            region,
            healthy: true,
        })
    }
}

#[tokio::main]
async fn main() -> PoolResult<()> {
    let pool = KeyedWorkerPool::new(
        RegionalConnectionFactory,
        KeyedWorkerPoolConfig::default()
            .with_max_count_per_key(Some(4))
            .with_max_idle_count_per_key(Some(2)),
    );

    let connection = pool.get_worker(Region::East).await?;
    assert_eq!(connection.region, Region::East);

    Ok(())
}

In a keyed pool, a worker created by the factory for key must satisfy both conditions:

worker.primary_key() == key
worker.supports(worker.primary_key()) == true

Otherwise, acquisition returns PoolErrorCode::InvalidConfig. A worker's primary key is cached when it is created. If that primary key changes while the worker is checked out, the worker is evicted when returned.

Keyed capacity semantics

max_count_per_key is an independent hard limit for each key. Once the limit is reached, new requests for that key wait for an existing worker. None leaves each key unlimited, while Some(0) causes requests to return an invalid-configuration error. There is no pool-wide worker-count limit, so activity for one key does not consume another key's capacity.

max_idle_count_per_key independently limits the idle cache for each primary key and counts neither checked-out nor pending workers. On return, a worker is handed directly to a compatible waiter when possible; otherwise it becomes the MRU worker in its primary-key bucket. When a bucket exceeds its limit, its oldest idle worker is evicted. None leaves each bucket unlimited; zero disables idle caching. There is no pool-wide idle-cache limit.

Explicit keyed acquisition only cleans and reuses the requested key's primary bucket. Workers from other primary-key buckets are not considered, even if worker.supports(requested_key) would return true.

Classified worker creation concurrency

ClassifiedWorkerPoolConfig limits concurrent calls to ClassifiedWorkerFactory::create to 20 by default. The limit is shared by generic get_worker() calls and classification-specific get_classified_worker() calls, preventing bursts of speculative generic creation from exhausting resources. Use with_max_concurrent_creation_count() to change it; zero is invalid and causes acquisition to return PoolErrorCode::InvalidConfig.

Clearing and errors

clear_all_worker().await performs the following operations:

Immediately removes all idle workers.
Completes currently queued requests with PoolErrorCode::Cleared.
Temporarily rejects new requests with PoolErrorCode::Clearing.
Waits for checked-out and currently creating workers to leave the pool.

The pool can be used again after clearing completes.

The public error types are:

PoolResult<T>: the common result type for pool operations
PoolError: an error containing a PoolErrorCode
PoolErrorCode::Failed: a general failure
PoolErrorCode::Clearing: the pool is currently being cleared
PoolErrorCode::Cleared: the request was terminated by a clear operation
PoolErrorCode::InvalidConfig: the configuration or a factory result is invalid

Implementation requirements

Worker::is_work(), KeyedWorker::is_work(), KeyedWorker::supports(), and KeyedWorker::primary_key() may be called while the pool's internal state lock is held. These methods must be fast and non-blocking, and they must not re-enter APIs on the same pool.

Idle timeouts do not start a background cleanup task. Explicit keyed or classified acquisition cleans only the requested bucket. A classification-free acquisition cleans each bucket it actually traverses while looking for a worker. Applications can call cleanup_idle_worker() to scan every idle bucket explicitly.

Development

cargo fmt -- --check
cargo check
cargo test

See doc/design.md for a more detailed discussion of scheduling and concurrency behavior.

License

This project is licensed under the terms described in the repository's LICENSE file.

sfo-pool 0.4.1