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