compio_pool/
lib.rs

1//! A connection pool for [`compio`], the completion-based, thread-per-core
2//! Rust runtime.
3//!
4//! # Why compio needs its own pool
5//!
6//! General-purpose pools (`bb8`, `deadpool`, `r2d2`) assume connections are
7//! `Send`: one shared `Mutex<Vec<Conn>>`, any worker takes any connection.
8//! `compio` breaks that assumption. Each thread runs its own driver
9//! (io_uring, IOCP or poll), IO handles are bound to the driver that created
10//! them, and buffers are handed to the kernel by ownership — so
11//! [`compio::net::TcpStream`] is `!Send` and cannot be put behind an `Arc<Mutex<_>>`
12//! at all.
13//!
14//! So "share one connection across threads simultaneously" is not the thing to
15//! build. Two different things hide behind that phrase:
16//!
17//! * **Concurrent use of a single connection** — only meaningful for protocols
18//!   that multiplex (HTTP/2, Redis pipelining). Otherwise a connection is an
19//!   exclusive resource and sharing means taking turns.
20//! * **A pool shared by every thread** — any thread can check out *some*
21//!   connection. That is what this crate provides.
22//!
23//! # Design
24//!
25//! [`Pool`] is `Send + Sync` and cheap to clone, but holds no connections. The
26//! connections live in a **per-thread shard** stored in a thread-local, so:
27//!
28//! * the acquire fast path touches no atomics and takes no lock;
29//! * [`Manage::Connection`] never needs to be `Send`;
30//! * a connection is always driven by the compio driver that created it;
31//! * at thread exit, that thread's connections close on their own thread.
32//!
33//! [`Config::max_size`] is therefore **per shard**, not per process. A global
34//! cap would need a cross-thread semaphore on the hot path, which is the exact
35//! contention thread-per-core exists to avoid.
36//!
37//! ## Letting connections migrate
38//!
39//! Pure sharding wastes connections when load is skewed: a quiet thread holds
40//! idle sockets a busy thread could use. The optional [`Reservoir`] fixes that
41//! with a bounded, lock-free `ArrayQueue` shared by every thread: a shard whose
42//! free list is over `min_idle` detaches the surplus into it, and a shard whose
43//! free list is empty pops from it before paying for a handshake. The claiming
44//! thread re-wraps the socket in its own runtime via [`Detach::attach`].
45//!
46//! It requires [`Detach`], which is **platform-dependent and deliberately
47//! opt-in**:
48//!
49//! | driver | `attach` | connections may change threads |
50//! |---|---|---|
51//! | io_uring (Linux) | no-op | yes |
52//! | poll (Unix fallback) | no-op | yes |
53//! | IOCP (Windows) | `CreateIoCompletionPort`, once per handle | **no** |
54//!
55//! ## Cancellation is the sharp edge
56//!
57//! With completion-based IO, dropping a future mid-operation does not undo the
58//! operation. The connection's protocol state is then unknown and it must be
59//! destroyed rather than returned. See [`Pooled::begin_op`].
60//!
61//! # Example
62//!
63//! ```no_run
64//! use std::time::Duration;
65//! use compio_pool::{Config, Manage, Pool, SlotMeta};
66//!
67//! struct Tcp(String);
68//!
69//! impl Manage for Tcp {
70//!     type Connection = compio::net::TcpStream;
71//!     type Error = std::io::Error;
72//!
73//!     async fn connect(&self) -> std::io::Result<Self::Connection> {
74//!         compio::net::TcpStream::connect(&self.0).await
75//!     }
76//!
77//!     async fn recycle(
78//!         &self,
79//!         conn: &mut Self::Connection,
80//!         _meta: &SlotMeta,
81//!     ) -> std::io::Result<()> {
82//!         conn.peer_addr().map(|_| ())
83//!     }
84//! }
85//!
86//! # async fn run() -> Result<(), Box<dyn std::error::Error>> {
87//! let pool = Pool::builder(Tcp("127.0.0.1:6379".into()))
88//!     .max_size(8)                                   // per thread
89//!     .min_idle(2)
90//!     .acquire_timeout(Duration::from_secs(5))
91//!     .build();
92//!
93//! // Clone `pool` onto every compio thread; each gets its own shard.
94//! let mut conn = pool.acquire().await?;
95//!
96//! let op = conn.begin_op();
97//! // ... use `conn` ...
98//! op.complete_op();
99//! # Ok(())
100//! # }
101//! ```
102
103#![warn(missing_docs)]
104
105mod config;
106mod error;
107mod exchange;
108mod guard;
109mod manage;
110mod metrics;
111mod pool;
112mod shard;
113mod slot;
114
115#[cfg(test)]
116mod test_support;
117
118pub use crate::{
119    config::Config,
120    error::Error,
121    exchange::{Exchange, NoExchange, Parked, Reservoir, Unparked},
122    guard::{OpGuard, Pooled},
123    manage::{Detach, Manage},
124    metrics::Metrics,
125    pool::{Builder, Pool},
126    slot::SlotMeta,
127};
compio_pool/lib.rs

compio_pool/
lib.rs
