Lesson 39 of 44
Performance and Tuning
Tuning changes a program to spend fewer resources per unit of work: fewer system calls, fewer wake-ups, fewer allocations, less time waiting for locks. Each lever in this lesson reduces one kind of operation, and each example counts that operation so the effect is visible without a stopwatch. Wall-clock times differ between machines and runs, so the lesson states trade-offs qualitatively and gives the criteria for choosing.
| Lever | What it reduces |
|---|---|
| Batching | wake-ups and system calls per item |
BufReader and BufWriter |
system calls per byte |
Sharding locks (dashmap) |
operations that compete for one lock |
Fewer spawn calls |
task allocations and scheduling hops |
watch and bounded channels |
queued work nobody needs |
Bytes |
copies of buffers |
| Worker count, per-core runtimes | cross-thread traffic |
yield_now, budget |
latency spikes for other tasks |
| Smaller futures | memory moved per spawn |
The examples use tokio, bytes, and dashmap (cargo add bytes dashmap).
Measure first
A change that is not measured is a guess. A workable order:
- Define the goal in one number: requests per second at a latency target (for example p99 below a limit), or memory per connection.
- Reproduce the load in a benchmark that runs the real code path, in a release build (
cargo build --release). Debug builds are many times slower and shift the bottleneck. - Record the baseline: throughput, latency percentiles (Tracing and Observability covers histograms), CPU use, task count, queue depths.
- Find where the time goes with a profiler (
perf, flamegraphs) and withtokio-consolefor tasks. Change one thing at a time. - Repeat the measurement. Keep the change only if the goal number improves.
Async code has its own failure signature: low CPU use with high latency means tasks are waiting (on locks, channels, or I/O), while one saturated core with low throughput means a task blocks or a lock is contended.
Example 1: Counting system calls with a counting writer
An unbuffered write_all of a small piece goes straight to the underlying stream. On a socket or file, each such call becomes a system call, a switch into the operating system. The example replaces the socket with a writer that only counts calls, so the number of writes is visible.
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::{AsyncWrite, AsyncWriteExt, BufWriter};
#[derive(Default)]
struct CountingWriter {
calls: usize,
bytes: usize,
}
impl AsyncWrite for CountingWriter {
fn poll_write(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
self.calls += 1;
self.bytes += buf.len();
Poll::Ready(Ok(buf.len()))
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
}
#[tokio::main]
async fn main() -> io::Result<()> {
// 10_000 small messages of 10 bytes each.
let mut direct = CountingWriter::default();
for _ in 0..10_000 {
direct.write_all(b"0123456789").await?;
}
println!("unbuffered: {} calls, {} bytes", direct.calls, direct.bytes);
let mut buffered = BufWriter::new(CountingWriter::default());
for _ in 0..10_000 {
buffered.write_all(b"0123456789").await?;
}
buffered.flush().await?;
let inner = buffered.get_ref();
println!("buffered: {} calls, {} bytes", inner.calls, inner.bytes);
Ok(())
}The same 100,000 bytes reach the writer in far fewer calls. BufWriter collects small writes in memory and passes them on when its buffer fills or when flush is called. The flush at the end is required: bytes still in the buffer are lost if the writer is dropped without it.
Example 2: Buffer size sets the call count
BufWriter::new uses a buffer of 8 KiB (DEFAULT_BUF_SIZE in the Tokio source is 8 * 1024). BufWriter::with_capacity picks another size. A larger buffer means fewer calls and more memory per connection. A smaller buffer means more calls and less memory.
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::{AsyncWrite, AsyncWriteExt, BufWriter};
#[derive(Default)]
struct CountingWriter {
calls: usize,
}
impl AsyncWrite for CountingWriter {
fn poll_write(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
self.calls += 1;
Poll::Ready(Ok(buf.len()))
}
fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
Poll::Ready(Ok(()))
}
}
#[tokio::main]
async fn main() -> io::Result<()> {
println!("{:>10} {:>8}", "capacity", "calls");
for capacity in [64, 512, 8 * 1024, 64 * 1024] {
let mut writer = BufWriter::with_capacity(capacity, CountingWriter::default());
for _ in 0..10_000 {
writer.write_all(b"0123456789").await?;
}
writer.flush().await?;
println!("{capacity:>10} {:>8}", writer.get_ref().calls);
}
Ok(())
}The call count falls roughly in proportion to the capacity, and the gain shrinks once a buffer is larger than typical bursts of data. Criteria for choosing a size: the size of one logical message or batch, the number of connections multiplied by two buffers each (memory), and the socket's own send buffer, since the operating system copies data in chunks of its own.
Example 3: Buffered reads
Reading a byte at a time from an unbuffered stream costs one read call per byte. BufReader reads a block from the inner stream and hands out pieces from memory. The example counts poll_read calls on an in-memory source.
use std::io;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, AsyncReadExt, BufReader, ReadBuf};
struct CountingReader {
data: Vec<u8>,
position: usize,
calls: usize,
}
impl CountingReader {
fn new(len: usize) -> Self {
CountingReader { data: vec![7; len], position: 0, calls: 0 }
}
}
impl AsyncRead for CountingReader {
fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
self.calls += 1;
let remaining = &self.data[self.position..];
let count = remaining.len().min(buf.remaining());
buf.put_slice(&remaining[..count]);
self.position += count;
Poll::Ready(Ok(()))
}
}
#[tokio::main]
async fn main() -> io::Result<()> {
let mut plain = CountingReader::new(20_000);
let mut sum = 0u64;
for _ in 0..20_000 {
sum += plain.read_u8().await? as u64;
}
println!("unbuffered read_u8: {} read calls (sum {sum})", plain.calls);
let mut buffered = BufReader::new(CountingReader::new(20_000));
let mut sum = 0u64;
for _ in 0..20_000 {
sum += buffered.read_u8().await? as u64;
}
println!("buffered read_u8: {} read calls (sum {sum})", buffered.get_ref().calls);
Ok(())
}Both loops read the same 20,000 bytes and compute the same sum. The buffered one asks the inner stream for a block of 8 KiB at a time. Reading in large chunks with read into a big buffer, or with read_buf into a BytesMut, gets the same benefit without BufReader; BufReader matters when the code needs many small reads such as read_u8, read_u32, or read_line.
Example 4: Batching messages with recv_many
Every recv().await on a channel is a separate poll of the consumer. When messages arrive in bursts, recv_many(&mut buffer, limit) moves up to limit queued messages into a Vec in one call and one wake-up. The example counts the calls needed to drain 100 queued messages.
use tokio::sync::mpsc;
#[tokio::main]
async fn main() {
let (tx, mut rx) = mpsc::channel::<u32>(128);
for n in 0..100 {
tx.send(n).await.unwrap();
}
let mut single_calls = 0;
for _ in 0..50 {
rx.recv().await.unwrap();
single_calls += 1;
}
println!("recv: {single_calls} calls for 50 messages");
let mut batch = Vec::new();
let mut batch_calls = 0;
let mut received = 0;
while received < 50 {
received += rx.recv_many(&mut batch, 16).await;
batch_calls += 1;
}
println!("recv_many: {batch_calls} calls for {received} messages");
}Batching also lets the consumer amortize per-batch work: one database transaction per batch, one write_all for many responses, one lock acquisition for many updates. The cost is latency: the first message of a batch waits for the batch to be taken, so batch size and any flush timer are set by the latency target.
Example 5: Counting wake-ups
A task is polled each time it is woken. If a producer wakes the consumer for every item, the consumer is polled once per item. The example wraps the consumer future in a small counter that counts polls, and compares a producer that sends one item at a time with one that sends the same items in groups.
use std::future::Future;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::sync::mpsc;
struct CountPolls<F> {
inner: Pin<Box<F>>,
polls: usize,
}
impl<F: Future> Future for CountPolls<F> {
type Output = (F::Output, usize);
fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
self.polls += 1;
match self.inner.as_mut().poll(cx) {
Poll::Ready(value) => Poll::Ready((value, self.polls)),
Poll::Pending => Poll::Pending,
}
}
}
async fn run(items_per_send: usize) -> usize {
let (tx, mut rx) = mpsc::channel::<Vec<u32>>(8);
tokio::spawn(async move {
for start in (0..120).step_by(items_per_send) {
let group: Vec<u32> = (start..start + items_per_send as u32).collect();
tx.send(group).await.unwrap();
tokio::task::yield_now().await; // give the consumer a turn after each send
}
});
let consumer = async move {
let mut total = 0;
while let Some(group) = rx.recv().await {
total += group.len();
}
total
};
let (total, polls) = CountPolls { inner: Box::pin(consumer), polls: 0 }.await;
assert_eq!(total, 120);
polls
}
#[tokio::main(flavor = "current_thread")]
async fn main() {
println!("1 item per send: consumer polled {} times", run(1).await);
println!("10 items per send: consumer polled {} times", run(10).await);
println!("40 items per send: consumer polled {} times", run(40).await);
}The number of polls follows the number of sends, not the number of items. Fewer, larger messages mean fewer wake-ups, which is the effect that recv_many, BufWriter, and message coalescing all rely on.
Example 6: Avoid a spawn that does not add parallelism
tokio::spawn allocates a task, moves the future into it, and schedules it. The spawned work can run in parallel on another worker thread and can outlive the caller. When neither is needed, tokio::join! runs several futures concurrently inside the current task with no allocation and no scheduling hop. The example reads the runtime's count of alive tasks in both variants.
use tokio::runtime::Handle;
use tokio::sync::oneshot;
async fn step(name: &str) -> String {
tokio::task::yield_now().await;
format!("{name} done")
}
#[tokio::main(flavor = "current_thread")]
async fn main() {
let metrics = Handle::current().metrics();
let (a, b, c) = tokio::join!(step("a"), step("b"), step("c"));
println!("join!: {a}, {b}, {c}; alive tasks while running: {}", metrics.num_alive_tasks());
let (gate_tx, gate_rx) = oneshot::channel::<()>();
let gate = std::sync::Arc::new(tokio::sync::Mutex::new(Some(gate_rx)));
let handles: Vec<_> = ["a", "b", "c"]
.into_iter()
.map(|name| {
let gate = gate.clone();
tokio::spawn(async move {
if let Some(rx) = gate.lock().await.take() {
let _ = rx.await;
}
step(name).await
})
})
.collect();
println!("spawn: alive tasks after 3 spawns: {}", metrics.num_alive_tasks());
let _ = gate_tx.send(());
for handle in handles {
handle.await.unwrap();
}
}The three join! branches share one task. The three spawns create three. Criteria for using spawn: the work should run in parallel on other threads, it should outlive the current function, it should be cancellable on its own (through its handle), or its failure must not take down the caller. Otherwise join!, select!, or a plain .await is cheaper. A spawn per tiny operation (one per element of a loop over a cheap computation) adds allocation and scheduling costs that exceed the work.
Example 7: When async is slower than threads
Async tasks pay for their concurrency with polling, waking, and state-machine bookkeeping. That cost is small next to waiting on a network, and large next to a tiny computation. The table gives the criteria; it contains no numbers because the numbers depend on the machine.
| Workload | Better fit | Reason |
|---|---|---|
| Thousands of mostly idle connections | async tasks | a task costs a small heap allocation, a thread costs a stack and a kernel object |
| Few connections, each doing heavy computation | threads or spawn_blocking |
no waiting to overlap; the scheduler adds overhead only |
| Short CPU-bound jobs on many items | rayon on its own thread pool |
data parallelism without polling overhead; bridge with spawn_blocking (Blocking and CPU-Bound Work) |
| Reading and writing local files | async wrapper is a convenience only | tokio::fs runs blocking calls on the blocking pool, so it adds a thread hop |
| Latency-critical path with one connection | either; measure | async adds a wake-up hop per await that would have blocked |
The rule that follows: async pays off when the work spends most of its time waiting, and waiting can be overlapped. A CPU-bound stage inside an async program belongs on the blocking pool or on a separate thread pool, or it blocks the workers that serve every other task.
Example 8: Lock contention and sharding
A lock is contended when several tasks try to hold it at once. One Mutex<HashMap> in front of all keys makes every operation compete for the same lock. Sharding splits the map into several maps, each with its own lock, and routes a key to one shard by its hash. Operations on different shards do not compete. The example counts how many operations each lock receives.
use std::collections::HashMap;
use std::collections::hash_map::DefaultHasher;
use std::hash::{Hash, Hasher};
use std::sync::Mutex;
use std::sync::atomic::{AtomicUsize, Ordering};
struct Shard {
map: Mutex<HashMap<u32, u64>>,
operations: AtomicUsize,
}
struct ShardedMap {
shards: Vec<Shard>,
}
impl ShardedMap {
fn new(count: usize) -> Self {
let shards = (0..count)
.map(|_| Shard { map: Mutex::new(HashMap::new()), operations: AtomicUsize::new(0) })
.collect();
ShardedMap { shards }
}
fn shard_for(&self, key: u32) -> &Shard {
let mut hasher = DefaultHasher::new();
key.hash(&mut hasher);
&self.shards[hasher.finish() as usize % self.shards.len()]
}
fn add(&self, key: u32, amount: u64) {
let shard = self.shard_for(key);
shard.operations.fetch_add(1, Ordering::Relaxed);
*shard.map.lock().unwrap().entry(key).or_insert(0) += amount;
}
}
fn main() {
for shard_count in [1, 4, 16] {
let map = ShardedMap::new(shard_count);
for key in 0..10_000u32 {
map.add(key, 1);
}
let busiest = map.shards.iter().map(|s| s.operations.load(Ordering::Relaxed)).max().unwrap();
let total: u64 = map.shards.iter().map(|s| s.map.lock().unwrap().values().sum::<u64>()).sum();
println!("{shard_count:>2} shard(s): busiest lock received {busiest} of 10000 operations, total {total}");
}
}With one shard, one lock receives every operation. With 16 shards, no lock receives much more than a sixteenth of them, so the chance that two tasks want the same lock at the same moment falls in proportion. Sharding does not help when most operations use the same key (a hot key), and it does not change the correctness rules for locks in async code: hold the lock briefly and never across an .await (Shared State: Mutex and RwLock).
Example 9: DashMap
The dashmap crate is a sharded concurrent hash map. Its methods take &self, so the map is shared through an Arc with no outer Mutex, and each operation locks only the shard that holds its key.
use dashmap::DashMap;
use std::sync::Arc;
#[tokio::main]
async fn main() {
let counts: Arc<DashMap<String, u64>> = Arc::new(DashMap::new());
let mut handles = Vec::new();
for worker in 0..8 {
let counts = counts.clone();
handles.push(tokio::spawn(async move {
for round in 0..1_000u32 {
let key = format!("key-{}", (worker + round) % 5);
*counts.entry(key).or_insert(0) += 1;
}
}));
}
for handle in handles {
handle.await.unwrap();
}
let mut entries: Vec<(String, u64)> = counts.iter().map(|e| (e.key().clone(), *e.value())).collect();
entries.sort();
println!("{entries:?}");
println!("total = {}", entries.iter().map(|(_, n)| n).sum::<u64>());
}entry(key).or_insert(0) locks the key's shard, inserts if missing, and returns a guard that dereferences to the value. The increment is atomic with respect to other operations on that shard, so no update is lost.
Example 10: Never hold a DashMap guard across .await
References returned by get, get_mut, and entry are guards that keep a shard locked until they are dropped. Holding one across an .await keeps the shard locked while the task is suspended. Other operations on that shard wait, and a second lock on the same shard from the same task deadlocks. The guards are also not Send, so tokio::spawn rejects such a future at compile time.
use dashmap::DashMap;
use std::sync::Arc;
#[tokio::main]
async fn main() {
let map: Arc<DashMap<u32, u32>> = Arc::new(DashMap::new());
map.insert(1, 10);
tokio::spawn(async move {
let guard = map.get(&1).unwrap();
tokio::task::yield_now().await;
println!("{}", *guard);
})
.await
.unwrap();
}The fix is to copy the data out and drop the guard before awaiting, for example let value = *map.get(&1).unwrap(); (the temporary guard is dropped at the end of the statement).
Example 11: watch drops stale work
A queue keeps every message. When the consumer only needs the latest state (a configuration, a status, a price), a queue makes it process obsolete values in order. A watch channel keeps one value and notifies receivers that it changed, so a slow receiver skips the values it missed (oneshot, broadcast, and watch).
use tokio::sync::{mpsc, watch};
#[tokio::main(flavor = "current_thread")]
async fn main() {
// A queue delivers all 1000 updates.
let (queue_tx, mut queue_rx) = mpsc::channel::<u32>(1000);
for value in 1..=1000 {
queue_tx.send(value).await.unwrap();
}
drop(queue_tx);
let mut queued = 0;
while queue_rx.recv().await.is_some() {
queued += 1;
}
println!("mpsc: consumer handled {queued} updates");
// A watch channel keeps only the latest one.
let (watch_tx, mut watch_rx) = watch::channel(0u32);
for value in 1..=1000 {
watch_tx.send(value).unwrap();
}
let mut observed = 0;
let mut last = 0;
while watch_rx.has_changed().unwrap() {
watch_rx.changed().await.unwrap();
last = *watch_rx.borrow_and_update();
observed += 1;
}
println!("watch: consumer handled {observed} update(s), last value {last}");
}Choosing a channel is a choice about what to lose under load. An unbounded mpsc grows without limit; a bounded mpsc makes senders wait; watch discards intermediate values; broadcast discards the oldest values for slow receivers. The right one depends on whether every message matters or only the latest state.
Example 12: Bytes avoids copying
Vec<u8>::clone copies the data. Bytes::clone increments a reference count and shares the same memory, and Bytes::slice creates a view of part of it. For data that is parsed, cached, or sent to several tasks, this removes copies. The example compares memory addresses; the addresses themselves are not printed.
use bytes::Bytes;
fn main() {
let original: Vec<u8> = (0..=255).collect();
let copy = original.clone();
println!("Vec clone shares memory: {}", original.as_ptr() == copy.as_ptr());
let shared = Bytes::from(original);
let clone = shared.clone();
println!("Bytes clone shares memory: {}", shared.as_ptr() == clone.as_ptr());
let header = shared.slice(0..16);
let body = shared.slice(16..);
println!("slice lengths: {} and {}", header.len(), body.len());
println!("slice points into the same buffer: {}", header.as_ptr() == shared.as_ptr());
println!("body starts 16 bytes later: {}", body.as_ptr() as usize - shared.as_ptr() as usize == 16);
}The shared memory is freed when the last Bytes handle drops. The consequence: a tiny slice of a large buffer keeps the whole buffer alive. Bytes::copy_from_slice makes a private copy when a small piece must outlive a large parent (Bytes and Framing).
Example 13: Worker thread count
The multi-thread runtime starts one worker thread per CPU core by default. Builder::worker_threads(n) or the worker_threads option of #[tokio::main] changes it. Handle::metrics().num_workers() reports the value in use.
use tokio::runtime::Builder;
fn main() {
for workers in [1, 2, 4] {
let runtime = Builder::new_multi_thread().worker_threads(workers).enable_all().build().unwrap();
let reported = runtime.block_on(async { tokio::runtime::Handle::current().metrics().num_workers() });
println!("configured {workers}, runtime reports {reported}");
}
let single = Builder::new_current_thread().enable_all().build().unwrap();
let reported = single.block_on(async { tokio::runtime::Handle::current().metrics().num_workers() });
println!("current_thread reports {reported}");
}Criteria for changing the default:
- Fewer workers than cores when the process shares the machine with other work (a container with a CPU quota, a co-located database), so the runtime does not compete for time slices.
- More workers than cores almost never helps, because workers run tasks and a CPU can run one thread at a time. Blocking work belongs on the separate blocking pool (
max_blocking_threads), not on more workers. - A single worker (or
current_thread) when tasks are light and cross-thread wake-ups cost more than parallelism saves.
Example 14: One current_thread runtime per core
The multi-thread scheduler moves tasks between workers, so data touched by a task may move between CPU caches and must be protected by Send types and locks. An alternative design runs one current_thread runtime per thread and routes work by key so each key always lands on the same thread. State that belongs to one shard needs no lock. This is the design of thread-per-core systems.
The example starts four shard threads. The main thread routes 1,000 keys to a shard by key % 4, and each shard counts what it received.
use tokio::sync::mpsc;
fn main() {
const SHARDS: usize = 4;
let mut senders = Vec::new();
let mut threads = Vec::new();
for shard in 0..SHARDS {
let (tx, mut rx) = mpsc::channel::<u32>(64);
senders.push(tx);
threads.push(std::thread::spawn(move || {
let runtime = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap();
runtime.block_on(async move {
let mut seen = 0usize;
let mut wrong_shard = 0usize;
while let Some(key) = rx.recv().await {
seen += 1;
if key as usize % SHARDS != shard {
wrong_shard += 1;
}
}
(shard, seen, wrong_shard)
})
}));
}
for key in 0..1_000u32 {
senders[key as usize % SHARDS].blocking_send(key).unwrap();
}
drop(senders);
for thread in threads {
let (shard, seen, wrong) = thread.join().unwrap();
println!("shard {shard}: {seen} keys, {wrong} routed incorrectly");
}
}Each thread owns its runtime, so nothing is stolen from it and its tasks stay on one core's cache. blocking_send is the Sender method for use from code that is not inside a runtime (mpsc Channels).
Pinning a thread to a CPU core (thread affinity) is a separate operating-system feature. The standard library has no API for it; crates such as core_affinity or tools such as taskset provide it. Pinning helps when cache locality matters and hurts when the operating system needs to move threads to balance load. It needs a measurement on the target hardware.
Example 15: LocalSet and Rc
A LocalSet runs tasks on the current thread only, so its tasks may be !Send: they can hold Rc and RefCell instead of Arc and Mutex. Rc counts references with plain integer operations, and Arc uses atomic operations. In a per-core design this removes locks and atomic counters from the hot path (Runtime Flavors and the Builder).
use std::cell::RefCell;
use std::rc::Rc;
use tokio::task::LocalSet;
#[tokio::main(flavor = "current_thread")]
async fn main() {
let local = LocalSet::new();
local
.run_until(async {
let counter = Rc::new(RefCell::new(0u32));
let mut handles = Vec::new();
for task in 0..4 {
let counter = Rc::clone(&counter);
handles.push(tokio::task::spawn_local(async move {
for _ in 0..250 {
*counter.borrow_mut() += 1;
tokio::task::yield_now().await;
}
task
}));
}
for handle in handles {
handle.await.unwrap();
}
println!("counter = {}", counter.borrow());
println!("strong count after tasks finished = {}", Rc::strong_count(&counter));
})
.await;
}The tasks interleave at each yield_now, yet the RefCell is never borrowed across an .await, so it never panics. The RefCell rule is the same as the lock rule: keep the borrow short and end it before awaiting.
Example 16: Yielding and the cooperative budget
Tokio gives each task a budget of 128 operations per poll (Budget::initial in task/coop/mod.rs). Tokio's own resources, such as channel receives and socket reads, spend one unit each time they return Ready. When the budget is used up, the next such operation returns Pending and wakes the task, which sends it to the back of the run queue and lets other tasks run. This bounds how long a task that always has data ready can hold a worker.
The example drains a full channel of 1,000 messages while a second task increments a counter each time it gets a turn. On the current_thread runtime the ordering is fixed. tokio::task::unconstrained turns the budget off for a future, which shows the effect.
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use tokio::sync::mpsc;
async fn drain(constrained: bool) -> u32 {
let (tx, mut rx) = mpsc::channel::<u32>(1000);
for n in 0..1000 {
tx.send(n).await.unwrap();
}
drop(tx);
let turns = Arc::new(AtomicU32::new(0));
let counter = turns.clone();
let ticker = tokio::spawn(async move {
loop {
counter.fetch_add(1, Ordering::SeqCst);
tokio::task::yield_now().await;
}
});
let consume = async move {
let mut received = 0;
while rx.recv().await.is_some() {
received += 1;
}
received
};
let received = if constrained { consume.await } else { tokio::task::unconstrained(consume).await };
ticker.abort();
assert_eq!(received, 1000);
turns.load(Ordering::SeqCst)
}
#[tokio::main(flavor = "current_thread")]
async fn main() {
println!("with the budget: other task got {} turns during the drain", drain(true).await);
println!("without the budget: other task got {} turns during the drain", drain(false).await);
}With the budget, the drain is interrupted every 128 messages so the other task runs. Without it, the receiving loop never returns Pending because data is always ready, and the other task waits until the loop ends. The budget only covers operations through Tokio's own primitives. A loop of plain computation never reaches it.
Example 17: yield_now in a CPU loop
A long loop with no .await holds its worker until it ends, because nothing in it gives the scheduler a chance to switch. Calling tokio::task::yield_now().await every N iterations returns control to the scheduler, which runs other ready tasks first. The number of yields is the loop length divided by N, so N trades responsiveness against the cost of each yield.
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
#[tokio::main(flavor = "current_thread")]
async fn main() {
let other_turns = Arc::new(AtomicU32::new(0));
let counter = other_turns.clone();
let other = tokio::spawn(async move {
loop {
counter.fetch_add(1, Ordering::SeqCst);
tokio::task::yield_now().await;
}
});
for every in [10_000usize, 1_000, 100] {
other_turns.store(0, Ordering::SeqCst);
let mut sum = 0u64;
let mut yields = 0;
for i in 0..100_000u64 {
sum += i * i % 7;
if (i as usize + 1) % every == 0 {
yields += 1;
tokio::task::yield_now().await;
}
}
println!(
"yield every {every:>5}: {yields:>4} yields, other task ran {} times (sum {sum})",
other_turns.load(Ordering::SeqCst)
);
}
other.abort();
}A yield that happens too rarely leaves other tasks waiting for the whole stretch. Yielding every iteration adds a scheduler round trip to each unit of work. For heavy computation, spawn_blocking avoids the question because the work leaves the async workers entirely.
Example 18: Future size
A spawned future is stored in the task's heap allocation, and every await in an async fn may keep locals alive inside the future. A large local held across an .await makes the future large, and moving a large future (into spawn, into join!, into a Box) copies it. size_of_val on a future reports its size.
use std::mem::size_of_val;
async fn small() {
tokio::task::yield_now().await;
}
async fn large_on_the_stack() {
let buffer = [0u8; 16 * 1024];
tokio::task::yield_now().await;
println!("{}", buffer.len());
}
async fn large_on_the_heap() {
let buffer = vec![0u8; 16 * 1024];
tokio::task::yield_now().await;
println!("{}", buffer.len());
}
async fn buffer_dropped_before_await() {
{
let buffer = [0u8; 16 * 1024];
println!("{}", buffer.len());
}
tokio::task::yield_now().await;
}
fn main() {
println!("small: {} bytes", size_of_val(&small()));
println!("array across await: {} bytes", size_of_val(&large_on_the_stack()));
println!("Vec across await: {} bytes", size_of_val(&large_on_the_heap()));
println!("array dropped earlier: {} bytes", size_of_val(&buffer_dropped_before_await()));
println!("array future is larger than 16 KiB: {}", size_of_val(&large_on_the_stack()) > 16 * 1024);
}The array lives inside the future because it is alive at the .await. The Vec keeps its data on the heap, so only a pointer, a capacity, and a length are stored. Ending the array's scope before the .await keeps it out of the future. The rule for shrinking futures: keep large values in Vec or Box, and drop them before awaiting when possible. Large futures are also the reason recursive async functions and deeply nested join! calls are boxed (Anatomy of a Task).
Example 19: A tuning checklist
Each row is a symptom, the measurement that confirms it, and the change that addresses it.
| Symptom | Confirm with | Change |
|---|---|---|
| High system-call rate per request | strace -c, perf |
BufReader and BufWriter, coalesce small writes |
| Consumer polled once per item | poll counts, tokio-console |
batch producers, recv_many |
| One core saturated, others idle | per-worker busy time (unstable metrics), CPU use | remove blocking calls, spawn_blocking, split the hot task |
| Latency spikes in unrelated requests | p99 histograms, poll-time metrics | yield_now in long loops, move CPU work off the workers |
| Time spent waiting for a lock | profiler, lock-hold spans | shorter critical sections, sharding, dashmap, message passing |
| Memory growth under load | queue-depth gauges, num_alive_tasks |
bounded channels, limits on spawned tasks (Backpressure and Load Shedding) |
| High allocation rate | heap profiler | Bytes, buffer reuse, BytesMut::split |
| Large per-task memory | size_of_val, memory per connection |
box large locals, shrink buffers |
Interview points
- Tune only against a measured goal, in a release build, changing one thing at a time.
- Async improves throughput when work waits and waiting overlaps. CPU-bound work does not benefit and belongs on
spawn_blockingorrayon. BufWriterandBufReadercut system calls; the default buffer is 8 KiB. Flush aBufWriterbefore drop.- Batching (
recv_many, coalesced writes) reduces wake-ups and per-item overhead and adds latency for the first item of a batch. - Spawn only when the work needs parallelism, a separate lifetime, or independent cancellation.
join!is a single task. - Contention drops with shorter critical sections, sharded locks,
dashmap, or ownership by one task with message passing. A hot key defeats sharding. - Never hold a
DashMapguard, astdlock guard, or aRefCellborrow across an.await. watchkeeps only the latest value; a queue keeps all values. The choice is about what may be lost under load.Bytes::cloneandsliceshare memory; a small slice keeps its whole parent buffer alive.- Worker count defaults to the core count. More workers than cores rarely helps; blocking work uses the blocking pool.
- One
current_threadruntime per core with key-based routing avoids locks and work stealing at the cost of load balancing. - The cooperative budget is 128 operations per poll and applies to Tokio's own primitives. Plain CPU loops need
yield_noworspawn_blocking. - Values held across an
.awaitlive in the future, so large locals make large futures.
Summary
- Measure with a release build, a stated goal, and percentiles.
- Count operations (calls, polls, allocations) to see the effect of a change independent of machine speed.
- Buffer small I/O, batch messages, and prefer fewer larger messages to many small ones.
- Reduce lock competition with shorter holds, sharding, or single-owner tasks.
- Use
spawnfor parallelism and independence,join!for concurrency inside one task. - Share buffers with
Bytes; keep the largest values out of futures. - Choose worker counts and runtime layout from the deployment: cores available, blocking work, and the cost of cross-thread wake-ups.
- Give other tasks a turn in long loops with
yield_now, or move the work tospawn_blocking.