Struct Pool 

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

A blueprint for per-thread pools: build it once, clone it onto each compio thread you spawn, and call local there.

Clone + Send + Sync. Cloning shares nothing that is touched on the hot path — only the manager, the Builder settings, the shutdown flag and the metrics registry. Every connection, idle list and slot counter belongs to one thread’s LocalPool.

Implementations

impl<M: ManageConnection> Pool<M>

pub fn builder() -> Builder<M>

Start configuring a pool.

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 max_size(&self) -> u32

The per-thread ceiling on open connections.

pub fn local(&self) -> LocalPool<M>

Make this thread’s pool.

Each call creates an independent LocalPool bound to the calling thread. Call it once per compio thread and keep the handle; it is Clone (a cheap Rc bump) if several tasks on that thread need it. Because the returned pool is !Send, the compiler guarantees it — and the ring-bound connections it holds — never move to another thread.

Examples found in repository

examples/tcp_pool.rs (line 133)

fn worker(core: cpu::CoreId, addr: SocketAddr, pool: Pool<Backend>) {
    cpu::pin_current(core);

    let runtime = Runtime::builder().build().expect("build compio runtime");
    runtime.block_on(async move {
        let local = pool.local();

        let std_listener = bind_reuseport(addr, 1024, None).expect("bind SO_REUSEPORT");
        let listener = TcpListener::from_std(std_listener).expect("wrap listener in ring");

        loop {
            let Ok((stream, _peer)) = listener.accept().await else {
                continue;
            };
            compio::runtime::spawn(serve(local.clone(), stream)).detach();
        }
    });
}

examples/wal.rs (line 139)

fn worker(core: cpu::CoreId, addr: SocketAddr, pool: Pool<WalManager>) {
    cpu::pin_current(core);

    let runtime = Runtime::builder().build().expect("build compio runtime");
    runtime.block_on(async move {
        let local = pool.local();

        let std_listener = bind_reuseport(addr, 1024, None).expect("bind SO_REUSEPORT");
        let listener = TcpListener::from_std(std_listener).expect("wrap listener in ring");

        loop {
            let Ok((stream, _peer)) = listener.accept().await else {
                continue;
            };
            compio::runtime::spawn(serve(local.clone(), stream)).detach();
        }
    });
}

examples/echo.rs (line 96)

fn worker(core: cpu::CoreId, addr: SocketAddr, pool: Pool<Buffers>) {
    cpu::pin_current(core);

    let runtime = Runtime::builder().build().expect("build compio runtime");
    runtime.block_on(async move {
        // This thread's pool. `!Send`, so it — and its ring-bound buffers —
        // cannot escape the thread.
        let local = pool.local();

        let std_listener = bind_reuseport(addr, 1024, None).expect("bind SO_REUSEPORT");
        let listener = TcpListener::from_std(std_listener).expect("wrap listener in ring");

        loop {
            let Ok((stream, _peer)) = listener.accept().await else {
                continue;
            };
            compio::runtime::spawn(serve(local.clone(), stream)).detach();
        }
    });
}

examples/file_server.rs (line 104)

fn worker(core: cpu::CoreId, addr: SocketAddr, pool: Pool<Buffers>) {
    cpu::pin_current(core);

    let runtime = Runtime::builder().build().expect("build compio runtime");
    runtime.block_on(async move {
        let local = pool.local();

        let std_listener = bind_reuseport(addr, 1024, None).expect("bind SO_REUSEPORT");
        let listener = TcpListener::from_std(std_listener).expect("wrap listener in ring");

        loop {
            let Ok((stream, _peer)) = listener.accept().await else {
                continue;
            };
            // The reply is a size header followed by the body — two writes. Without
            // TCP_NODELAY, Nagle holds the body waiting for the header's ACK, which
            // the peer delays ~40 ms, so disable it on every accepted connection.
            let _ = stream.set_nodelay(true);
            compio::runtime::spawn(serve(local.clone(), stream)).detach();
        }
    });
}

examples/task_echo.rs (line 73)

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();
        }
    })
}

examples/unix_echo.rs (line 76)

fn main() -> io::Result<()> {
    let path = std::env::args()
        .nth(1)
        .unwrap_or_else(|| "/tmp/compio-pool-echo.sock".into());
    let capacity: u32 = std::env::args()
        .nth(2)
        .and_then(|s| s.parse().ok())
        .unwrap_or(1024);

    // A stale socket file from a previous run would make `bind` fail with
    // EADDRINUSE, so clear it first.
    let _ = std::fs::remove_file(&path);

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

    // One listener, so one worker. Pin it to the first permitted core so the ring
    // and the socket share a CPU, the same discipline the fan-out examples apply
    // per thread.
    if let Some(core) = cpu::cores().into_iter().next() {
        cpu::pin_current(core);
    }
    println!("unix_echo on {path}: 1 worker, capacity {capacity}");

    let runtime = Runtime::builder().build().expect("build compio runtime");
    runtime.block_on(async move {
        let local = pool.local();
        let listener = UnixListener::bind(&path).await?;

        loop {
            let Ok((stream, _peer)) = listener.accept().await else {
                continue;
            };
            compio::runtime::spawn(serve(local.clone(), stream)).detach();
        }
    })
}

Additional examples can be found in:

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

pub fn close(&self)

Shut the pool down for every thread.

After this, each thread’s LocalPool::get fails immediately with RunError::Closed, a returned PooledConnection is dropped instead of recycled, and warm / maintain become no-ops. Connections already checked out keep working until their holders drop them.

The flag is shared, but the wakers that unpark a task waiting in get are not — they live on each worker’s own thread. A task already parked waiting for a slot therefore unblocks on its next local wake (another connection returning, or LocalPool::clear) or when its connection_timeout elapses, not the instant close is called from another thread. For a prompt drain, stop each worker’s accept loop and call LocalPool::clear on its thread.

pub fn is_closed(&self) -> bool

Whether close has been called.

pub fn state(&self) -> State

A process-wide snapshot: the live connection, idle and waiter counts summed across every thread’s LocalPool. Takes the registry lock, so call it off the hot path (a metrics scrape, a health endpoint). For one thread’s numbers use LocalPool::state.

pub fn statistics(&self) -> Statistics

Process-wide cumulative counters summed across every thread’s LocalPool. Like state it takes the registry lock and is meant for a scrape, not the hot path. For one thread’s totals use LocalPool::statistics.

Trait Implementations

impl<M: ManageConnection> Clone for Pool<M>

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl<M: ManageConnection> Debug for Pool<M>

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.