Struct LocalPool 

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

One thread’s pool, created by Pool::local.

Holds the idle connections and the slot count for this thread alone. Clone shares the same per-thread pool between tasks on the thread; it is never Send.

Implementations

impl<M: ManageConnection> LocalPool<M>

pub fn max_size(&self) -> u32

The per-thread ceiling on open connections.

pub fn state(&self) -> State

A snapshot of this thread’s open, idle and waiting counts. Lock-free. For the process-wide sum use Pool::state.

Examples found in repository

examples/warmup.rs (line 82)

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

examples/fanout_requests.rs (line 142)

fn main() -> io::Result<()> {
    let requests: usize = arg(1).unwrap_or(50_000);
    let capacity: u32 = arg(2).unwrap_or(8);
    let clients: usize = arg(3).unwrap_or(128);
    let size: usize = arg(4).unwrap_or(256);

    let runtime = Runtime::builder().build().expect("build compio runtime");
    runtime.block_on(async move {
        // 1. An in-process echo backend on an ephemeral port, run as a task.
        let listen: SocketAddr = "127.0.0.1:0".parse().unwrap();
        let backend = TcpListener::bind(listen).await?;
        let backend_addr = backend.local_addr()?;
        compio::runtime::spawn(backend_loop(backend)).detach();

        // 2. One pool of upstreams to the backend, and this thread's LocalPool.
        let pool = Pool::builder()
            .max_size(capacity)
            .build(Upstreams { addr: backend_addr });
        let local = pool.local();

        println!(
            "fanout_requests -> {backend_addr}: {clients} client tasks over {capacity} pooled \
             upstreams, {requests} requests x {size} B"
        );

        // Shared, single-thread state: plain Rc<Cell>, no atomics needed.
        let next = Rc::new(Cell::new(0usize)); // next request index to claim
        let bytes = Rc::new(Cell::new(0u64)); // total bytes echoed back
        let inflight = Rc::new(Cell::new(0u32)); // upstreams in use right now
        let peak = Rc::new(Cell::new(0u32)); // high-water mark of `inflight`

        // 3. Fan out: spawn the client tasks and await them all.
        let started = Instant::now();
        let mut tasks = Vec::with_capacity(clients);
        for _ in 0..clients {
            tasks.push(compio::runtime::spawn(client(
                local.clone(),
                next.clone(),
                requests,
                size,
                bytes.clone(),
                inflight.clone(),
                peak.clone(),
            )));
        }
        for task in tasks {
            let _ = task.await;
        }
        let elapsed = started.elapsed();

        println!(
            "done: {requests} requests in {:.3}s ({:.0} req/s), {} B echoed",
            elapsed.as_secs_f64(),
            requests as f64 / elapsed.as_secs_f64(),
            bytes.get(),
        );
        println!(
            "pool: opened {} upstreams total, peak {} in use at once (cap {capacity}); final {:?}",
            OPENS.load(Ordering::Relaxed),
            peak.get(),
            local.state(),
        );
        Ok(())
    })
}

pub fn statistics(&self) -> Statistics

This thread’s cumulative counters. Lock-free. For the process-wide sum use Pool::statistics.

pub async fn warm(&self) -> Result<u32, RunError<M::Error>>

Open connections up to min_idle and park them on the idle list. Returns how many were created. Tops up to the floor each time it is called, so it doubles as the warm-up and the min-idle maintenance step; maintain pairs it with a reap. A no-op (returns Ok(0)) once the pool is closed. Stops early and returns RunError::User if the manager fails to connect.

Examples found in repository

examples/warmup.rs (line 85)

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

Drop idle connections that have passed idle_timeout or max_lifetime, freeing their slots. Returns how many were reaped.

Expiry is otherwise lazy — a connection is only checked as it is handed out or returned — so a pool that goes quiet would hold dead backend connections open indefinitely. Call this on a timer (see maintain) to retire them promptly. Only idle connections are considered; one that is checked out is left to its holder.

pub async fn maintain(&self) -> Result<u32, RunError<M::Error>>

The periodic maintenance step: reap expired idle connections, then warm back up to min_idle. Returns how many fresh connections were opened. Call it from a timer on each worker thread — thread-per-core has no shared reaper thread, so maintenance is yours to drive:

use std::time::Duration;
compio::runtime::spawn(async move {
    loop {
        compio::time::sleep(Duration::from_secs(30)).await;
        let _ = local.maintain().await;
    }
})
.detach();

pub fn clear(&self) -> u32

Drop every idle connection now, freeing their slots, and return how many were dropped. Connections currently checked out are untouched and return as usual. Use it to drain after a backend failover, or as part of a shutdown once Pool::close has been called. Unlike reap it ignores the timeouts and clears unconditionally.

pub async fn get(&self) -> Result<PooledConnection<M>, RunError<M::Error>>

Check a connection out of this thread’s pool, waiting up to connection_timeout for one.

Reuses an idle connection when there is one (validated first if test_on_check_out is set), otherwise opens one while under max_size, otherwise waits for a slot. The returned PooledConnection derefs to the connection and returns it to this pool when dropped. Fails with RunError::Closed if the pool has been closed.

Examples found in repository

examples/scatter_gather.rs (line 222)

async fn query_shard(pool: LocalPool<Shard>, query: String) -> Option<String> {
    let mut conn = match pool.get().await {
        Ok(conn) => conn,
        Err(_) => return None,
    };
    if conn.write_line(&query).await.is_err() {
        conn.healthy = false;
        return None;
    }
    match conn.read_line().await {
        Ok(Some(reply)) => Some(reply),
        _ => {
            conn.healthy = false;
            None
        }
    }
}

examples/echo.rs (line 113)

async fn serve(local: LocalPool<Buffers>, mut stream: TcpStream) {
    let mut lease = match local.get().await {
        Ok(lease) => lease,
        Err(_) => return,
    };
    loop {
        let mut buf = std::mem::take(&mut *lease);
        buf.clear();
        let BufResult(read, buf) = stream.read(buf).await;
        match read {
            Ok(0) | Err(_) => {
                *lease = buf;
                return;
            }
            Ok(_) => {}
        }
        let BufResult(written, buf) = stream.write_all(buf).await;
        *lease = buf;
        if written.is_err() {
            return;
        }
    }
}

examples/task_echo.rs (line 92)

async fn serve(local: LocalPool<Buffers>, mut stream: TcpStream) {
    let mut lease = match local.get().await {
        Ok(lease) => lease,
        Err(_) => return,
    };
    loop {
        let mut buf = std::mem::take(&mut *lease);
        buf.clear();
        let BufResult(read, buf) = stream.read(buf).await;
        match read {
            Ok(0) | Err(_) => {
                *lease = buf;
                return;
            }
            Ok(_) => {}
        }
        let BufResult(written, buf) = stream.write_all(buf).await;
        *lease = buf;
        if written.is_err() {
            return;
        }
    }
}

examples/unix_echo.rs (line 90)

async fn serve(local: LocalPool<Buffers>, mut stream: UnixStream) {
    let mut lease = match local.get().await {
        Ok(lease) => lease,
        Err(_) => return,
    };
    loop {
        let mut buf = std::mem::take(&mut *lease);
        buf.clear();
        let BufResult(read, buf) = stream.read(buf).await;
        match read {
            Ok(0) | Err(_) => {
                *lease = buf;
                return;
            }
            Ok(_) => {}
        }
        let BufResult(written, buf) = stream.write_all(buf).await;
        *lease = buf;
        if written.is_err() {
            return;
        }
    }
}

examples/wal.rs (line 156)

async fn serve(local: LocalPool<WalManager>, mut stream: TcpStream) {
    let mut lease = match local.get().await {
        Ok(lease) => lease,
        Err(_) => return,
    };
    let mut buf = Vec::with_capacity(64 * 1024);
    loop {
        buf.clear();
        let BufResult(read, b) = stream.read(buf).await;
        buf = b;
        let n = match read {
            Ok(0) | Err(_) => return,
            Ok(n) => n,
        };

        // Append the record at the segment tail. Read `offset` into a local so the
        // expression does not borrow `lease` twice.
        let offset = lease.offset;
        let BufResult(w, b) = lease.file.write_all_at(buf, offset).await;
        buf = b;
        if w.is_err() {
            return;
        }
        lease.offset += n as u64;

        if *SYNC && lease.file.sync_data().await.is_err() {
            return;
        }

        let BufResult(w, _) = stream.write_all(b"OK\n".to_vec()).await;
        if w.is_err() {
            return;
        }
    }
}

examples/tcp_pool.rs (line 161)

async fn serve(local: LocalPool<Backend>, mut client: TcpStream) {
    let mut buf = Vec::with_capacity(16 * 1024);
    loop {
        buf.clear();
        let BufResult(read, b) = client.read(buf).await;
        buf = b;
        let n = match read {
            Ok(0) | Err(_) => return,
            Ok(n) => n,
        };

        // Lease one upstream for just this round trip.
        let mut up = match local.get().await {
            Ok(up) => up,
            Err(_) => return, // pool exhausted past the timeout, or connect failed
        };

        // Request out to the upstream.
        let BufResult(written, b) = up.stream.write_all(buf).await;
        buf = b;
        if written.is_err() {
            up.healthy = false;
            return;
        }

        // Echo backend replies with exactly what it got, so read n back.
        let BufResult(got, reply) = up.stream.read_exact(Vec::with_capacity(n)).await;
        if got.is_err() {
            up.healthy = false;
            return;
        }
        drop(up); // return the upstream to the pool before touching the client

        let BufResult(written, _) = client.write_all(reply).await;
        if written.is_err() {
            return;
        }
    }
}

Additional examples can be found in:

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

Trait Implementations

impl<M: ManageConnection> Clone for LocalPool<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 LocalPool<M>

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

Formats the value using the given formatter.