rust-impl-async-tokio
The tokio runtime implementation skill: how to start a tokio runtime, spawn and join async tasks, race futures with select!, apply timeouts, cancel work safely, and use JoinSet for structured concurrency. Targets tokio 1.x on Rust 1.85+, edition 2024.
Cross-references: [[rust-core-async-runtime]] (Future trait, Waker model, why a runtime is needed), [[rust-syntax-async-await]] (async fn, .await, Send/Sync bounds across .await), [[rust-impl-channels]] (tokio::sync::mpsc, oneshot, broadcast, watch), [[rust-errors-async]] (JoinError, Elapsed, propagating panics across task boundaries).
When to use this skill
- User adds
tokiotoCargo.tomland asks which features (rt,rt-multi-thread,macros,time,sync,full). - User annotates
mainwith#[tokio::main]or hand-builds aRuntime. - User calls
tokio::spawn, getsJoinHandle<T>, asks how to wait for it or abort it. - User has CPU-bound or blocking sync work inside async code and asks where to run it.
- User wants to race two futures or apply a timeout.
- User wants to spawn a group of related tasks and cancel them together (structured concurrency).
- User sees
panicked at 'there is no reactor running'orcannot block the current thread from within a runtime.
For the Future trait, pinning, and runtime architecture see [[rust-core-async-runtime]]. For async fn / .await syntax and Send-across-await see [[rust-syntax-async-await]]. For channels see [[rust-impl-channels]]. For errors see [[rust-errors-async]].
Decision tree: which runtime flavor
Single-threaded app, !Send data held across .await, lowest overhead?
-> Runtime::new_current_thread() or #[tokio::main(flavor = "current_thread")]
Multi-core throughput, work-stealing across threads, default for servers?
-> Runtime::new_multi_thread() or #[tokio::main] (default)
Embedded inside another runtime / async-std / smol?
-> NEVER build a nested tokio runtime; use Handle::current() to schedule.
ALWAYS pick current_thread when tasks hold Rc, RefCell, or any !Send type across .await. NEVER pick multi_thread and then sprinkle Arc<Mutex<_>> everywhere to silence Send errors; switch flavor instead.
Decision tree: where to run blocking code
Is the operation < 100 microseconds and non-IO?
YES -> inline async, no spawn needed.
NO -> continue.
Is it CPU-bound (parse, compress, crypto, math)?
YES -> rayon thread pool OR tokio::task::spawn_blocking (small ops only)
NO -> continue.
Is it synchronous IO with no async equivalent (legacy driver, FFI)?
YES -> tokio::task::spawn_blocking (returns JoinHandle)
OR tokio::task::block_in_place (multi_thread only, current task tagged)
Is an async equivalent available (tokio::fs, tokio::net, reqwest, sqlx)?
YES -> use the async API. NEVER call std::fs or std::net inside async fn.
ALWAYS prefer spawn_blocking over block_in_place when you do not need to keep the task identity. NEVER call std::thread::sleep, std::sync::Mutex::lock() across .await, or synchronous reqwest::blocking::* inside async code: each one parks a worker thread and starves every other task on that worker.
Runtime setup
The #[tokio::main] macro is sugar that desugars to a Runtime::new() plus block_on(async { ... }). Pick the explicit form when you need to configure threads, worker stack size, or runtime hooks.
Sugared form
// Cargo.toml needs: tokio = { version = "1", features = ["macros", "rt-multi-thread"] }
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let body = reqwest::get("https://example.com").await?.text().await?;
println!("{}", body);
Ok(())
}
ALWAYS enable the macros feature for #[tokio::main]. ALWAYS enable rt-multi-thread (default flavor) or rt (current_thread flavor); without them the macro panics at startup.
Single-threaded variant
#[tokio::main(flavor = "current_thread")]
async fn main() {
let local = std::rc::Rc::new(42); // !Send, fine on current_thread
do_work(local.clone()).await;
}
Explicit form (custom thread count, hooks, worker stack)
fn main() -> std::io::Result<()> {
let runtime = tokio::runtime::Builder::new_multi_thread()
.worker_threads(4)
.thread_name("workers")
.thread_stack_size(2 * 1024 * 1024)
.enable_all() // time + io drivers
.build()?;
runtime.block_on(async {
// your async work
});
Ok(())
}
NEVER call Runtime::new() from inside an async fn running on tokio: it panics with Cannot start a runtime from within a runtime. Use Handle::current() to schedule onto the existing one. See references/methods.md for Handle APIs.
Spawning tasks
tokio::spawn(future) returns a JoinHandle<T>. The future starts running immediately on the runtime; you do not need to .await the handle for the task to make progress.
let handle: tokio::task::JoinHandle<u32> = tokio::spawn(async {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
42
});
let value = handle.await?; // returns Result<T, JoinError>
Send bounds:
multi_threadruntime: the future must beSend + 'static. The compiler checks every type held across.await.current_threadruntime: the future must be'staticbut notSend.
JoinError carries one of three causes: the task panicked, the task was cancelled (.abort() or drop), or it was deliberately aborted with AbortHandle. ALWAYS match on err.is_panic() vs err.is_cancelled(); never assume ? propagates a meaningful error message. Full method list in references/methods.md.
Detaching
Dropping a JoinHandle does NOT abort the task on tokio. The task keeps running until it finishes or the runtime shuts down; its return value is discarded. ALWAYS keep the handle when you need the result; ALWAYS hold an AbortHandle if you need remote cancellation without owning the handle.
Aborting
let handle = tokio::spawn(long_running());
handle.abort(); // schedules cancellation, returns immediately
let result = handle.await; // Err(JoinError) with is_cancelled() == true
Abort is cooperative: tokio cancels the task at the next .await point. Sync code between two .await points runs to completion. NEVER rely on .abort() to stop CPU-bound code that never yields; use spawn_blocking plus an Arc<AtomicBool> cancellation flag instead.
Selecting between futures: tokio::select!
select! polls multiple futures concurrently and returns as soon as one completes. The other branches are cancelled (their futures are dropped without re-polling).
tokio::select! {
msg = rx.recv() => handle(msg),
_ = tokio::time::sleep(Duration::from_secs(5)) => timeout(),
_ = shutdown.cancelled() => shutdown_handler(),
}
Branches are polled in random order by default to avoid starvation. ALWAYS pass biased; as the first token when you need deterministic priority (biased; makes branches poll top-to-bottom).
tokio::select! {
biased;
_ = shutdown.cancelled() => return, // always check shutdown first
msg = rx.recv() => handle(msg),
}
Cancellation safety: any future inside select! may be cancelled at any .await. NEVER put a future that drops important state mid-step inside select! unless that future is documented as cancellation-safe. tokio::sync::mpsc::Receiver::recv and tokio::time::sleep are safe; many third-party futures are not. See references/anti-patterns.md.
Optional pre-condition (if guard):
tokio::select! {
msg = rx.recv(), if !paused => handle(msg),
_ = wakeup.notified() => paused = false,
}
Optional else branch fires when every branch's if-guard is false:
tokio::select! {
msg = rx.recv(), if !paused => handle(msg),
else => return,
}
Timeouts
tokio::time::timeout(duration, future) returns Result<T, Elapsed>. The inner future is dropped (cancelled) when the deadline expires.
use tokio::time::{timeout, Duration};
match timeout(Duration::from_secs(3), client.fetch(url)).await {
Ok(Ok(body)) => process(body),
Ok(Err(e)) => log_error(e), // inner future returned Err
Err(_elapsed) => log_timeout(), // deadline hit
}
For a fixed wall-clock deadline use tokio::time::timeout_at(Instant, future). NEVER wrap a non-cancellation-safe future in timeout: the inner future will be dropped mid-step and may leak resources, double-close, or split a multi-step transaction. Wrap such futures in their own task and use a CancellationToken + select! instead.
Cancellation semantics
Tokio cancellation is drop-based: when a future is dropped, every .await chain above it sees no more progress. There is no panic, no exception. Destructors run as normal.
Three primary cancellation triggers:
JoinHandle::abort()schedules the task for cancellation at the next.await.select!drops the losing branches.timeoutdrops the inner future on deadline.
To make your code cancellation-tolerant: keep .await points cheap, use tokio_util::sync::CancellationToken for cooperative shutdown signals, and use Drop impls or scopeguard::defer! for cleanup that must run even when cancelled.
let token = tokio_util::sync::CancellationToken::new();
let child = token.child_token();
tokio::spawn(async move {
tokio::select! {
_ = child.cancelled() => cleanup(),
_ = work_loop() => {}
}
});
// elsewhere:
token.cancel(); // all child tokens fire
ALWAYS prefer CancellationToken over AtomicBool for async cancellation: cancelled() is awaitable and integrates with select!. NEVER busy-loop on an atomic flag inside async code.
Structured concurrency with JoinSet
JoinSet<T> owns a group of spawned tasks, gives unordered join_next() access to whichever finishes first, and aborts every task on drop. This is the structured-concurrency primitive in tokio (NDST 12).
use tokio::task::JoinSet;
let mut set: JoinSet<u32> = JoinSet::new();
for n in 0..10 {
set.spawn(async move { do_work(n).await });
}
while let Some(res) = set.join_next().await {
match res {
Ok(value) => collect(value),
Err(e) if e.is_panic() => std::panic::resume_unwind(e.into_panic()),
Err(e) => log_cancellation(e),
}
}
Key methods:
spawn(future)(multi_thread) /spawn_local(future)(LocalSet).join_next() -> Option<Result<T, JoinError>>returns the next finished task orNonewhen empty.join_next_with_id()also returns the taskId.abort_all()aborts every member; pending tasks becomeis_cancelled()errors.shutdown().awaitaborts every task and waits for them to finish.detach_all()removes every task without aborting (the tasks keep running).
ALWAYS use JoinSet::shutdown().await in cleanup paths so cancelled tasks have a chance to run their destructors before the set is dropped. ALWAYS propagate panics with resume_unwind if you want a panicking child to terminate the parent. NEVER swallow JoinError::Panic silently; structured concurrency means children's failures are visible.
For a worked example (parallel HTTP fan-out with bounded concurrency, panic propagation, graceful shutdown) see references/examples.md.
Task-local storage: task_local!
tokio::task_local! {
static REQUEST_ID: String;
}
REQUEST_ID.scope("req-123".to_string(), async {
REQUEST_ID.with(|id| println!("{}", id));
}).await;
Each task sees its own value. Inherited automatically by child tasks only when spawned from inside the scope. ALWAYS use task_local! for tracing contexts, request IDs, or per-request configuration. NEVER use thread_local! for per-task data: tokio worker threads run many tasks and the value bleeds between them.
Quick checklist
#[tokio::main]withmacros+rt-multi-threadfeatures inCargo.toml.tokio::spawnfor async work,spawn_blockingfor sync work,block_in_placeonly on multi_thread when the task identity must stay.- Use
JoinSetwhen spawning more than two related tasks. ALWAYS join or abort on shutdown. - Wrap external futures in
timeoutonly when they are cancellation-safe. select!withbiased;for priority; default polling order is random.- Hold
std::sync::Mutexonly across non-.awaitregions; usetokio::sync::Mutexotherwise. - Cancellation is drop; place cleanup in
Dropimpls orscopeguard::defer!.
Reference index
references/methods.md:Runtime/Handle/JoinHandle/AbortHandle/JoinSetAPI surface with signatures.references/examples.md: worked patterns (graceful shutdown, fan-out with bounded concurrency, timeout-with-retry, cancellation tokens, task-local tracing).references/anti-patterns.md: ten field-tested anti-patterns with root cause and the correct fix.