Struct Builder 

pub struct Builder<M: ManageConnection> { /* private fields */ }

Configures and builds a Pool.

Every size here is per thread: max_size caps one thread’s LocalPool, not the process. A 48-core server with max_size(16) can therefore hold up to 48 × 16 connections, 16 of them to any one thread’s ring.

Defaults match bb8: max_size 10, min_idle 0, connection_timeout 30s, idle_timeout 10 min, max_lifetime 30 min, test_on_check_out on.

Implementations

impl<M: ManageConnection> Builder<M>

pub fn new() -> Self

A builder with the default settings.

pub fn max_size(self, n: u32) -> Self

The most connections one thread’s pool keeps open at once (idle plus checked out). Default 10.

Examples found in repository

examples/udp_echo.rs (line 72)

fn main() -> io::Result<()> {
    let addr: SocketAddr = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "0.0.0.0:7000".into())
        .parse()
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
    let capacity: u32 = std::env::args()
        .nth(2)
        .and_then(|s| s.parse().ok())
        .unwrap_or(1024);

    let pool = Pool::builder().max_size(capacity).build(Buffers);

    let cores = cpu::cores();
    if cores.is_empty() {
        return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
    }
    println!(
        "udp_echo on {addr}: {} workers, capacity {capacity}/worker",
        cores.len()
    );

    let mut handles = Vec::new();
    for (index, core) in cores.into_iter().enumerate() {
        let pool = pool.clone();
        let handle = thread::Builder::new()
            .name(format!("worker/{index}"))
            .spawn(move || worker(core, addr, pool))?;
        handles.push(handle);
    }
    for handle in handles {
        let _ = handle.join();
    }
    Ok(())
}

examples/file_server.rs (line 73)

fn main() -> io::Result<()> {
    let addr: SocketAddr = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "0.0.0.0:7000".into())
        .parse()
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
    let capacity: u32 = std::env::args()
        .nth(2)
        .and_then(|s| s.parse().ok())
        .unwrap_or(1024);

    let pool = Pool::builder().max_size(capacity).build(Buffers);

    let cores = cpu::cores();
    if cores.is_empty() {
        return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
    }
    println!(
        "file_server on {addr} serving {}: {} workers, capacity {capacity}/worker",
        BASE_DIR.display(),
        cores.len()
    );

    let mut handles = Vec::new();
    for (index, core) in cores.into_iter().enumerate() {
        let pool = pool.clone();
        let handle = thread::Builder::new()
            .name(format!("worker/{index}"))
            .spawn(move || worker(core, addr, pool))?;
        handles.push(handle);
    }
    for handle in handles {
        let _ = handle.join();
    }
    Ok(())
}

examples/echo.rs (line 62)

fn main() -> io::Result<()> {
    let addr: SocketAddr = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "0.0.0.0:7000".into())
        .parse()
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
    let capacity: u32 = std::env::args()
        .nth(2)
        .and_then(|s| s.parse().ok())
        .unwrap_or(1024);

    // Build the pool once. `Pool` is `Send + Clone`; every worker gets a clone.
    let pool = Pool::builder().max_size(capacity).build(Buffers);

    let cores = cpu::cores();
    if cores.is_empty() {
        return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
    }
    println!(
        "echo on {addr}: {} workers, capacity {capacity}/worker",
        cores.len()
    );

    let mut handles = Vec::new();
    for (index, core) in cores.into_iter().enumerate() {
        let pool = pool.clone();
        let handle = thread::Builder::new()
            .name(format!("worker/{index}"))
            .spawn(move || worker(core, addr, pool))?;
        handles.push(handle);
    }
    for handle in handles {
        let _ = handle.join();
    }
    Ok(())
}

examples/tcp_pool.rs (line 100)

fn main() -> io::Result<()> {
    let addr: SocketAddr = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "0.0.0.0:7000".into())
        .parse()
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
    let capacity: u32 = std::env::args()
        .nth(2)
        .and_then(|s| s.parse().ok())
        .unwrap_or(64);

    // Build the pool once; every worker gets a clone of the `Send` handle.
    let pool = Pool::builder().max_size(capacity).build(Backend);

    let cores = cpu::cores();
    if cores.is_empty() {
        return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
    }
    println!(
        "tcp_pool on {addr} -> upstream {}: {} workers, {capacity} upstreams/worker",
        *UPSTREAM,
        cores.len()
    );

    let mut handles = Vec::new();
    for (index, core) in cores.into_iter().enumerate() {
        let pool = pool.clone();
        let handle = thread::Builder::new()
            .name(format!("worker/{index}"))
            .spawn(move || worker(core, addr, pool))?;
        handles.push(handle);
    }
    for handle in handles {
        let _ = handle.join();
    }
    Ok(())
}

examples/wal.rs (line 107)

fn main() -> io::Result<()> {
    let addr: SocketAddr = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "0.0.0.0:7000".into())
        .parse()
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
    let capacity: u32 = std::env::args()
        .nth(2)
        .and_then(|s| s.parse().ok())
        .unwrap_or(1024);

    std::fs::create_dir_all(&*WAL_DIR)?;
    let pool = Pool::builder().max_size(capacity).build(WalManager);

    let cores = cpu::cores();
    if cores.is_empty() {
        return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
    }
    println!(
        "wal on {addr} -> {} (fdatasync: {}): {} workers, capacity {capacity}/worker",
        WAL_DIR.display(),
        if *SYNC { "on" } else { "off" },
        cores.len()
    );

    let mut handles = Vec::new();
    for (index, core) in cores.into_iter().enumerate() {
        let pool = pool.clone();
        let handle = thread::Builder::new()
            .name(format!("worker/{index}"))
            .spawn(move || worker(core, addr, pool))?;
        handles.push(handle);
    }
    for handle in handles {
        let _ = handle.join();
    }
    Ok(())
}

examples/task_echo.rs (line 67)

fn main() -> io::Result<()> {
    let addr: SocketAddr = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "127.0.0.1:7000".into())
        .parse()
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
    let capacity: u32 = std::env::args()
        .nth(2)
        .and_then(|s| s.parse().ok())
        .unwrap_or(1024);

    let pool = Pool::builder().max_size(capacity).build(Buffers);

    // One runtime, one ring. Everything below runs on this single thread.
    let runtime = Runtime::builder().build().expect("build compio runtime");
    runtime.block_on(async move {
        // This runtime's pool. `!Send`, so it and its buffers stay on this thread.
        let local = pool.local();

        let listener = TcpListener::bind(addr).await?;
        println!("task_echo on {addr}: 1 runtime, capacity {capacity}");

        // Accept forever; hand each connection to its own task on this same ring.
        loop {
            let Ok((stream, _peer)) = listener.accept().await else {
                continue;
            };
            let _ = stream.set_nodelay(true);
            compio::runtime::spawn(serve(local.clone(), stream)).detach();
        }
    })
}

Additional examples can be found in:

• examples/unix_echo.rs
• examples/warmup.rs
• examples/http_proxy.rs
• examples/scatter_gather.rs
• examples/fd_pipe.rs
• examples/fanout_requests.rs

pub fn min_idle(self, n: u32) -> Self

How many connections each thread opens up front and keeps idle, created by LocalPool::warm. Capped at max_size. Default 0.

Examples found in repository

examples/warmup.rs (line 76)

fn main() -> io::Result<()> {
    // Keep at most 8 per thread; warm 4 of them ahead of demand.
    let pool = Pool::builder().max_size(8).min_idle(4).build(Manager);

    // One runtime, one thread — the pool's natural unit.
    let runtime = Runtime::builder().build().expect("build compio runtime");
    runtime.block_on(async move {
        let local = pool.local();
        println!("fresh pool:     {:?}  opened={}", local.state(), opened());

        // Warm opens min_idle connections up front and parks them on the idle list.
        let made = local.warm().await.expect("warm");
        println!(
            "after warm:     {:?}  opened={}  (warm made {made})",
            local.state(),
            opened()
        );

        // Checking three out pulls them off the warm idle list — no new opens.
        let a = local.get().await.expect("get a");
        let b = local.get().await.expect("get b");
        let c = local.get().await.expect("get c");
        println!("3 checked out:  {:?}  opened={}", local.state(), opened());

        // Dropping the leases returns the resources to the idle list.
        drop((a, b, c));
        println!("3 returned:     {:?}  opened={}", local.state(), opened());

        // A burst reuses the 4 idle resources first, then opens more — but only up
        // to max_size=8, which caps the pool. A 9th get here would wait for a free
        // slot rather than open a ninth.
        let mut held = Vec::new();
        for _ in 0..8 {
            held.push(local.get().await.expect("burst"));
        }
        println!("8 checked out:  {:?}  opened={}", local.state(), opened());

        drop(held);
        println!("8 returned:     {:?}  opened={}", local.state(), opened());
    });

    Ok(())
}

pub fn connection_timeout(self, d: Duration) -> Self

How long get waits — for a free slot and a live connection — before returning RunError::TimedOut. Default 30s.

pub fn idle_timeout(self, d: Option<Duration>) -> Self

Drop an idle connection that has gone unused for this long. None keeps idle connections forever. Default 10 minutes.

pub fn max_lifetime(self, d: Option<Duration>) -> Self

Drop a connection older than this, idle or not, the next time it is checked out or returned. None never retires on age. Default 30 minutes.

pub fn test_on_check_out(self, yes: bool) -> Self

Call ManageConnection::is_valid on a reused connection as it is checked out. Default true.

pub fn build(self, manager: M) -> Pool<M>

Finish, producing the Send + Clone Pool blueprint.

Panics

If max_size is 0.

Examples found in repository

examples/udp_echo.rs (line 72)

fn main() -> io::Result<()> {
    let addr: SocketAddr = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "0.0.0.0:7000".into())
        .parse()
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
    let capacity: u32 = std::env::args()
        .nth(2)
        .and_then(|s| s.parse().ok())
        .unwrap_or(1024);

    let pool = Pool::builder().max_size(capacity).build(Buffers);

    let cores = cpu::cores();
    if cores.is_empty() {
        return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
    }
    println!(
        "udp_echo on {addr}: {} workers, capacity {capacity}/worker",
        cores.len()
    );

    let mut handles = Vec::new();
    for (index, core) in cores.into_iter().enumerate() {
        let pool = pool.clone();
        let handle = thread::Builder::new()
            .name(format!("worker/{index}"))
            .spawn(move || worker(core, addr, pool))?;
        handles.push(handle);
    }
    for handle in handles {
        let _ = handle.join();
    }
    Ok(())
}

examples/file_server.rs (line 73)

fn main() -> io::Result<()> {
    let addr: SocketAddr = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "0.0.0.0:7000".into())
        .parse()
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
    let capacity: u32 = std::env::args()
        .nth(2)
        .and_then(|s| s.parse().ok())
        .unwrap_or(1024);

    let pool = Pool::builder().max_size(capacity).build(Buffers);

    let cores = cpu::cores();
    if cores.is_empty() {
        return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
    }
    println!(
        "file_server on {addr} serving {}: {} workers, capacity {capacity}/worker",
        BASE_DIR.display(),
        cores.len()
    );

    let mut handles = Vec::new();
    for (index, core) in cores.into_iter().enumerate() {
        let pool = pool.clone();
        let handle = thread::Builder::new()
            .name(format!("worker/{index}"))
            .spawn(move || worker(core, addr, pool))?;
        handles.push(handle);
    }
    for handle in handles {
        let _ = handle.join();
    }
    Ok(())
}

examples/echo.rs (line 62)

fn main() -> io::Result<()> {
    let addr: SocketAddr = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "0.0.0.0:7000".into())
        .parse()
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
    let capacity: u32 = std::env::args()
        .nth(2)
        .and_then(|s| s.parse().ok())
        .unwrap_or(1024);

    // Build the pool once. `Pool` is `Send + Clone`; every worker gets a clone.
    let pool = Pool::builder().max_size(capacity).build(Buffers);

    let cores = cpu::cores();
    if cores.is_empty() {
        return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
    }
    println!(
        "echo on {addr}: {} workers, capacity {capacity}/worker",
        cores.len()
    );

    let mut handles = Vec::new();
    for (index, core) in cores.into_iter().enumerate() {
        let pool = pool.clone();
        let handle = thread::Builder::new()
            .name(format!("worker/{index}"))
            .spawn(move || worker(core, addr, pool))?;
        handles.push(handle);
    }
    for handle in handles {
        let _ = handle.join();
    }
    Ok(())
}

examples/tcp_pool.rs (line 100)

fn main() -> io::Result<()> {
    let addr: SocketAddr = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "0.0.0.0:7000".into())
        .parse()
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
    let capacity: u32 = std::env::args()
        .nth(2)
        .and_then(|s| s.parse().ok())
        .unwrap_or(64);

    // Build the pool once; every worker gets a clone of the `Send` handle.
    let pool = Pool::builder().max_size(capacity).build(Backend);

    let cores = cpu::cores();
    if cores.is_empty() {
        return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
    }
    println!(
        "tcp_pool on {addr} -> upstream {}: {} workers, {capacity} upstreams/worker",
        *UPSTREAM,
        cores.len()
    );

    let mut handles = Vec::new();
    for (index, core) in cores.into_iter().enumerate() {
        let pool = pool.clone();
        let handle = thread::Builder::new()
            .name(format!("worker/{index}"))
            .spawn(move || worker(core, addr, pool))?;
        handles.push(handle);
    }
    for handle in handles {
        let _ = handle.join();
    }
    Ok(())
}

examples/wal.rs (line 107)

fn main() -> io::Result<()> {
    let addr: SocketAddr = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "0.0.0.0:7000".into())
        .parse()
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
    let capacity: u32 = std::env::args()
        .nth(2)
        .and_then(|s| s.parse().ok())
        .unwrap_or(1024);

    std::fs::create_dir_all(&*WAL_DIR)?;
    let pool = Pool::builder().max_size(capacity).build(WalManager);

    let cores = cpu::cores();
    if cores.is_empty() {
        return Err(io::Error::new(io::ErrorKind::Unsupported, "no cores to run on"));
    }
    println!(
        "wal on {addr} -> {} (fdatasync: {}): {} workers, capacity {capacity}/worker",
        WAL_DIR.display(),
        if *SYNC { "on" } else { "off" },
        cores.len()
    );

    let mut handles = Vec::new();
    for (index, core) in cores.into_iter().enumerate() {
        let pool = pool.clone();
        let handle = thread::Builder::new()
            .name(format!("worker/{index}"))
            .spawn(move || worker(core, addr, pool))?;
        handles.push(handle);
    }
    for handle in handles {
        let _ = handle.join();
    }
    Ok(())
}

examples/task_echo.rs (line 67)

fn main() -> io::Result<()> {
    let addr: SocketAddr = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "127.0.0.1:7000".into())
        .parse()
        .map_err(|e| io::Error::new(io::ErrorKind::InvalidInput, format!("bad address: {e}")))?;
    let capacity: u32 = std::env::args()
        .nth(2)
        .and_then(|s| s.parse().ok())
        .unwrap_or(1024);

    let pool = Pool::builder().max_size(capacity).build(Buffers);

    // One runtime, one ring. Everything below runs on this single thread.
    let runtime = Runtime::builder().build().expect("build compio runtime");
    runtime.block_on(async move {
        // This runtime's pool. `!Send`, so it and its buffers stay on this thread.
        let local = pool.local();

        let listener = TcpListener::bind(addr).await?;
        println!("task_echo on {addr}: 1 runtime, capacity {capacity}");

        // Accept forever; hand each connection to its own task on this same ring.
        loop {
            let Ok((stream, _peer)) = listener.accept().await else {
                continue;
            };
            let _ = stream.set_nodelay(true);
            compio::runtime::spawn(serve(local.clone(), stream)).detach();
        }
    })
}

Additional examples can be found in:

• examples/unix_echo.rs
• examples/warmup.rs
• examples/http_proxy.rs
• examples/scatter_gather.rs
• examples/fd_pipe.rs
• examples/fanout_requests.rs

Trait Implementations

impl<M: ManageConnection> Default for Builder<M>

fn default() -> Self

Returns the “default value” for a type.