use async_trait::async_trait;
use std::io;
#[async_trait]
pub trait ConnectionFactory: Send + Sync + 'static {
type Connection: Send + 'static;
async fn connect(&self) -> io::Result<Self::Connection>;
}
pub struct TcpFactory {
pub addr: String,
}
#[async_trait]
impl ConnectionFactory for TcpFactory {
type Connection = tokio::net::TcpStream;
async fn connect(&self) -> io::Result<Self::Connection> {
tokio::net::TcpStream::connect(&self.addr).await
}
}
use crate::factory::ConnectionFactory;
use crate::guard::PooledConnection;
use std::sync::Arc;
use tokio::sync::{Mutex, Semaphore};
pub struct Pool<F: ConnectionFactory> {
factory: Arc<F>,
idle: Arc<Mutex<Vec<F::Connection>>>,
permits: Arc<Semaphore>,
}
impl<F: ConnectionFactory> Pool<F> {
pub fn new(factory: F, max_size: usize) -> Self {
assert!(max_size > 0, "pool size must be positive");
Pool {
factory: Arc::new(factory),
idle: Arc::new(Mutex::new(Vec::with_capacity(max_size))),
permits: Arc::new(Semaphore::new(max_size)),
}
}
pub async fn acquire(&self) -> std::io::Result<PooledConnection<F>> {
let permit = self
.permits
.clone()
.acquire_owned()
.await
.expect("semaphore closed");
let existing = self.idle.lock().await.pop();
let conn = match existing {
Some(c) => c,
None => self.factory.connect().await?,
};
Ok(PooledConnection::new(conn, self.idle.clone(), permit))
}
pub fn available_permits(&self) -> usize {
self.permits.available_permits()
}
}
use crate::factory::ConnectionFactory;
use std::ops::{Deref, DerefMut};
use std::sync::Arc;
use tokio::sync::{Mutex, OwnedSemaphorePermit};
pub struct PooledConnection<F: ConnectionFactory> {
conn: Option<F::Connection>,
idle: Arc<Mutex<Vec<F::Connection>>>,
_permit: OwnedSemaphorePermit,
}
impl<F: ConnectionFactory> PooledConnection<F> {
pub(crate) fn new(
conn: F::Connection,
idle: Arc<Mutex<Vec<F::Connection>>>,
permit: OwnedSemaphorePermit,
) -> Self {
PooledConnection {
conn: Some(conn),
idle,
_permit: permit,
}
}
}
impl<F: ConnectionFactory> Deref for PooledConnection<F> {
type Target = F::Connection;
fn deref(&self) -> &Self::Target {
self.conn.as_ref().expect("connection taken")
}
}
impl<F: ConnectionFactory> DerefMut for PooledConnection<F> {
fn deref_mut(&mut self) -> &mut Self::Target {
self.conn.as_mut().expect("connection taken")
}
}
impl<F: ConnectionFactory> Drop for PooledConnection<F> {
fn drop(&mut self) {
if let Some(conn) = self.conn.take() {
// Drop is sync; take the lock without awaiting.
if let Ok(mut idle) = self.idle.try_lock() {
idle.push(conn);
}
}
// _permit drops here, freeing a slot for the next waiter.
}
}
This snippet shows a small, self-contained async connection pool built on tokio, the kind of primitive that sits under higher-level database crates. The core idea is that a pool bounds the number of live connections and hands them out one at a time, blocking (asynchronously) when the pool is exhausted so callers naturally experience backpressure instead of opening an unbounded number of sockets.
In pool.rs, the Pool owns a Semaphore sized to max_size and a Mutex<Vec<C>> of idle connections. The semaphore is the real gatekeeper: acquire first takes a permit via acquire_owned, and only once a permit is held does it try to pop an idle connection or build a fresh one through the ConnectionFactory. Because the permit is owned (OwnedSemaphorePermit), it can live inside the guard for as long as the caller holds the connection, which is what keeps the in-flight count correct. The permit count and the vector of idle connections stay in sync without any spin-looping.
The returned PooledConnection in guard.rs is an RAII handle. It derefs to the underlying connection so callers use it transparently, and its Drop implementation returns the connection to the idle list and drops the permit, freeing a slot for the next waiter. The try_lock fallback in Drop matters: Drop cannot be async, so it uses the blocking lock and simply discards the connection if the mutex is momentarily contended, which is safe because losing a pooled connection only forces a later rebuild.
The ConnectionFactory trait in factory.rs decouples the pool from any specific driver; a Postgres, Redis, or mock factory all implement the same async connect. This is the extension point that makes the pool reusable.
The main trade-off is simplicity over features: there is no health-checking, idle timeout, or connection aging here, so a broken connection could be reused. The design assumes callers hold guards briefly. It fits cases where an app needs strict concurrency limits and predictable resource usage without pulling in a full pooling library.
Related snips
export interface RetryOptions {
retries: number;
baseMs: number;
maxMs: number;
signal?: AbortSignal;
onRetry?: (attempt: number, delay: number, err: unknown) => void;Exponential backoff with jitter for retries
struct Config<'a> {
name: &'a str,
value: &'a str,
}
fn parse_config(line: &str) -> Config {Lifetime annotations for flexible borrowing in structs
use crossbeam::channel::unbounded;
use std::thread;
fn main() {
let (tx, rx) = unbounded();Crossbeam for advanced concurrent data structures
// Creating a Promise
const myPromise = new Promise((resolve, reject) => {
const success = true;
setTimeout(() => {
if (success) {Promises and async/await patterns for asynchronous JavaScript
use tracing::{info, instrument};
#[instrument]
fn process_request(user_id: u64) {
info!(user_id, "Processing request");
// Work happens heretracing for structured logging and distributed tracing
use std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
Share this code
Here's the card — post it anywhere.
