use tokio::sync::broadcast;
pub struct Shutdown {
is_shutdown: bool,
notify: broadcast::Receiver<()>,
}
impl Shutdown {
pub fn new(notify: broadcast::Receiver<()>) -> Shutdown {
Shutdown {
is_shutdown: false,
notify,
}
}
pub fn is_shutdown(&self) -> bool {
self.is_shutdown
}
pub async fn recv(&mut self) {
if self.is_shutdown {
return;
}
// A closed sender also counts as a shutdown request.
let _ = self.notify.recv().await;
self.is_shutdown = true;
}
}
use crate::shutdown::Shutdown;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::broadcast;
use tokio_util::task::TaskTracker;
pub struct Listener {
pub listener: TcpListener,
pub notify_shutdown: broadcast::Sender<()>,
pub tracker: TaskTracker,
}
struct Handler {
stream: TcpStream,
shutdown: Shutdown,
}
impl Listener {
pub async fn run(&mut self) -> std::io::Result<()> {
loop {
let (stream, addr) = self.listener.accept().await?;
tracing::info!(%addr, "accepted connection");
let mut handler = Handler {
stream,
shutdown: Shutdown::new(self.notify_shutdown.subscribe()),
};
self.tracker.spawn(async move {
if let Err(err) = handler.run().await {
tracing::warn!(?err, "connection error");
}
});
}
}
}
impl Handler {
async fn run(&mut self) -> std::io::Result<()> {
let mut buf = [0u8; 1024];
while !self.shutdown.is_shutdown() {
let n = tokio::select! {
res = self.stream.read(&mut buf) => res?,
_ = self.shutdown.recv() => {
self.stream.write_all(b"server shutting down\n").await?;
return Ok(());
}
};
if n == 0 {
return Ok(()); // client closed the socket
}
self.stream.write_all(&buf[..n]).await?;
}
Ok(())
}
}
mod server;
mod shutdown;
use server::Listener;
use tokio::net::TcpListener;
use tokio::signal;
use tokio::sync::broadcast;
use tokio_util::task::TaskTracker;
#[tokio::main]
async fn main() -> std::io::Result<()> {
tracing_subscriber::fmt::init();
let listener = TcpListener::bind("127.0.0.1:6379").await?;
let (notify_shutdown, _) = broadcast::channel::<()>(1);
let tracker = TaskTracker::new();
let mut server = Listener {
listener,
notify_shutdown: notify_shutdown.clone(),
tracker: tracker.clone(),
};
tokio::select! {
res = server.run() => {
if let Err(err) = res {
tracing::error!(?err, "accept loop failed");
}
}
_ = signal::ctrl_c() => {
tracing::info!("ctrl-c received, draining connections");
}
}
// Dropping every sender broadcasts the shutdown to live handlers.
drop(notify_shutdown);
drop(server);
tracker.close();
tracker.wait().await;
tracing::info!("shutdown complete");
Ok(())
}
Graceful shutdown is the discipline of stopping a server without dropping in-flight work: the listener stops accepting new connections, existing connections are given a chance to finish, and the process exits only once everything has drained. This snippet wires that behavior into a tokio TCP server using a broadcast channel as a fan-out cancellation signal plus a TaskTracker to wait for outstanding tasks.
In shutdown.rs, the Shutdown type wraps a single-value broadcast::Receiver<()>. Every connection handler holds its own clone-derived receiver, and Shutdown::recv awaits the signal exactly once, latching is_shutdown so subsequent calls return immediately. A broadcast channel is chosen over a one-shot because the same shutdown notification must reach an arbitrary number of live connections simultaneously; sending on the paired Sender wakes them all. The notified flag guards against the common pitfall of awaiting an already-closed channel in a loop.
In server.rs, Listener::run owns the TcpListener, the broadcast::Sender used to hand out receivers, and a TaskTracker that records every spawned handler. The core is a tokio::select! that races listener.accept() against notify_shutdown in the handler; when a connection arrives, a new Handler is spawned onto the tracker with its own Shutdown::new(notify_shutdown.subscribe()). Each Handler::run again uses select! so a blocking read on the socket is interruptible — if the shutdown fires mid-read, the branch cancels the pending future and the handler returns cleanly rather than hanging.
In main.rs, signal::ctrl_c is awaited alongside the server via select!. When Ctrl-C lands, the server future is dropped (halting accept), then dropping the notify_shutdown sender broadcasts the signal, tracker.close() marks the tracker complete, and tracker.wait() blocks until every in-flight Handler finishes. This ordering matters: closing the tracker before all tasks are spawned would let wait() return early. The trade-off is that truly stuck handlers can still stall exit, so production code usually wraps the final wait() in a tokio::time::timeout to bound the drain window. This pattern generalizes to any accept-loop service where clean draining beats an abrupt process::exit.
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
class Product(models.Model):
name = models.CharField(max_length=200)
slug = models.SlugField(blank=True)
price = models.DecimalField(max_digits=10, decimal_places=2)
cost = models.DecimalField(max_digits=10, decimal_places=2)
margin = models.DecimalField(max_digits=5, decimal_places=2, blank=True)Django model signals vs overriding save
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 std::sync::mpsc;
use std::thread;
fn main() {
let (tx, rx) = mpsc::channel();Channels (mpsc) for message passing between threads
export type Settled<R> =
| { status: 'fulfilled'; value: R }
| { status: 'rejected'; reason: unknown };
export interface ConcurrencyOptions {
limit: number;Simple concurrency limiter for batch operations
Share this code
Here's the card — post it anywhere.
