tokio-patterns
Common Tokio patterns and idioms for async programming. Use when implementing worker pools, request-response patterns, pub/sub, timeouts, retries, or graceful shutdown.
- Installs
- 117
- Last 24h
- —
- Trending rank
- —
- Security
- Not audited
Install
Works with every agent the skills CLI supports.
npx skills add https://github.com/geoffjay/claude-plugins --skill tokio-patterns- 01
Install
Run the command in your project. The CLI asks which agents to install for.
- 02
Describe the task
Ask your agent for something this skill covers. It loads tokio-patterns automatically when relevant.
- 03
Review the result
The agent follows the SKILL.md steps, scripts and references bundled with the skill.
SKILL.md
The instructions your agent reads when the skill is triggered.
Tokio Patterns
This skill provides common patterns and idioms for building robust async applications with Tokio.
Worker Pool Pattern
Limit concurrent task execution using a semaphore:
use tokio::sync::Semaphore;
use std::sync::Arc;
pub struct WorkerPool {
semaphore: Arc<Semaphore>,
}
impl WorkerPool {
pub fn new(size: usize) -> Self {
Self {
semaphore: Arc::new(Semaphore::new(size)),
}
}
pub async fn execute<F, T>(&self, f: F) -> T
where
F: Future<Output = T>,
{
let _permit = self.semaphore.acquire().await.unwrap();
f.await
}
}
// Usage
let pool = WorkerPool::new(10);
let results = futures::future::join_all(
(0..100).map(|i| pool.execute(process_item(i)))
).await;
Request-Response Pattern
Use oneshot channels for request-response communication:
use tokio::sync::{mpsc, oneshot};
pub enum Command {
Get { key: String, respond_to: oneshot::Sender<Option<String>> },
Set { key: String, value: String },
}
pub async fn actor(mut rx: mpsc::Receiver<Command>) {
let mut store = HashMap::new();
while let Some(cmd) = rx.recv().await {
match cmd {
Command::Get { key, respond_to } => {
let value = store.get(&key).cloned();
let _ = respond_to.send(value);
}
Command::Set { key, value } => {
store.insert(key, value);
}
}
}
}
// Client usage
let (tx, rx) = mpsc::channel(32);
tokio::spawn(actor(rx));
let (respond_to, response) = oneshot::channel();
tx.send(Command::Get { key: "foo".into(), respond_to }).await.unwrap();
let value = response.await.unwrap();
Pub/Sub with Channels
Use broadcast channels for pub/sub messaging:
use tokio::sync::broadcast;
pub struct PubSub<T: Clone> {
tx: broadcast::Sender<T>,
}
impl<T: Clone> PubSub<T> {
pub fn new(capacity: usize) -> Self {
let (tx, _) = broadcast::channel(capacity);
Self { tx }
}
pub fn subscribe(&self) -> broadcast::Receiver<T> {
self.tx.subscribe()
}
pub fn publish(&self, message: T) -> Result<usize, broadcast::error::SendError<T>> {
self.tx.send(message)
}
}
// Usage
let pubsub = PubSub::new(100);
// Subscriber 1
let mut rx1 = pubsub.subscribe();
tokio::spawn(async move {
while let Ok(msg) = rx1.recv().await {
println!("Subscriber 1: {:?}", msg);
}
});
// Subscriber 2
let mut rx2 = pubsub.subscribe();
tokio::spawn(async move {
while let Ok(msg) = rx2.recv().await {
println!("Subscriber 2: {:?}", msg);
}
});
// Publisher
pubsub.publish("Hello".to_string()).unwrap();
Timeout Pattern
Wrap operations with timeouts:
use tokio::time::{timeout, Duration};
pub async fn with_timeout<F, T>(duration: Duration, future: F) -> Result<T, TimeoutError>
where
F: Future<Output = Result<T, Error>>,
{
match timeout(duration, future).await {
Ok(Ok(result)) => Ok(result),
Ok(Err(e)) => Err(TimeoutError::Inner(e)),
Err(_) => Err(TimeoutError::Elapsed),
}
}
// Usage
let result = with_timeout(
Duration::from_secs(5),
fetch_data()
).await?;
Retry with Exponential Backoff
Retry failed operations with backoff:
use tokio::time::{sleep, Duration};
pub async fn retry_with_backoff<F, T, E>(
mut operation: F,
max_retries: u32,
initial_backoff: Duration,
) -> Result<T, E>
where
F: FnMut() -> Pin<Box<dyn Future<Output = Result<T, E>>>>,
{
let mut retries = 0;
let mut backoff = initial_backoff;
loop {
match operation().await {
Ok(result) => return Ok(result),
Err(e) if retries < max_retries => {
retries += 1;
sleep(backoff).await;
backoff *= 2; // Exponential backoff
}
Err(e) => return Err(e),
}
}
}
// Usage
let result = retry_with_backoff(
|| Box::pin(fetch_data()),
3,
Duration::from_millis(100)
).await?;
Graceful Shutdown
Coordinate graceful shutdown across components:
use tokio::sync::broadcast;
use tokio::select;
pub struct ShutdownCoordinator {
tx: broadcast::Sender<()>,
}
impl ShutdownCoordinator {
pub fn new() -> Self {
let (tx, _) = broadcast::channel(1);
Self { tx }
}
pub fn subscribe(&self) -> broadcast::Receiver<()> {
self.tx.subscribe()
}
pub fn shutdown(&self) {
let _ = self.tx.send(());
}
}
// Worker pattern
pub async fn worker(mut shutdown: broadcast::Receiver<()>) {
loop {
select! {
_ = shutdown.recv() => {
// Cleanup
break;
}
result = do_work() => {
// Process result
}
}
}
}
// Main
let coordinator = ShutdownCoordinator::new();
let shutdown_rx1 = coordinator.subscribe();
let h1 = tokio::spawn(worker(shutdown_rx1));
let shutdown_rx2 = coordinator.subscribe();
let h2 = tokio::spawn(worker(shutdown_rx2));
// Wait for signal
tokio::signal::ctrl_c().await.unwrap();
coordinator.shutdown();
// Wait for workers
let _ = tokio::join!(h1, h2);
Async Initialization
Lazy async initialization with OnceCell:
use tokio::sync::OnceCell;
pub struct Service {
connection: OnceCell<Connection>,
}
impl Service {
pub fn new() -> Self {
Self {
connection: OnceCell::new(),
}
}
async fn get_connection(&self) -> &Connection {
self.connection
.get_or_init(|| async {
Connection::connect().await.unwrap()
})
.await
}
pub async fn query(&self, sql: &str) -> Result<Vec<Row>> {
let conn = self.get_connection().await;
conn.query(sql).await
}
}
Resource Cleanup with Drop
Ensure cleanup even on task cancellation:
pub struct Resource {
handle: SomeHandle,
}
impl Resource {
pub async fn new() -> Self {
Self {
handle: acquire_resource().await,
}
}
pub async fn use_resource(&self) -> Result<()> {
// Use the resource
Ok(())
}
}
impl Drop for Resource {
fn drop(&mut self) {
// Synchronous cleanup
// For async cleanup, use a separate shutdown method
self.handle.close();
}
}
// For async cleanup
impl Resource {
pub async fn shutdown(self) {
// Async cleanup
self.handle.close_async().await;
}
}
Select Multiple Futures
Use select! to race multiple operations:
use tokio::select;
pub async fn select_example() {
let mut rx1 = channel1();
let mut rx2 = channel2();
loop {
select! {
msg = rx1.recv() => {
if let Some(msg) = msg {
handle_channel1(msg).await;
} else {
break;
}
}
msg = rx2.recv() => {
if let Some(msg) = msg {
handle_channel2(msg).await;
} else {
break;
}
}
_ = tokio::time::sleep(Duration::from_secs(60)) => {
check_timeout().await;
}
}
}
}
Cancellation Token Pattern
Use tokio_util::sync::CancellationToken for cooperative cancellation:
use tokio_util::sync::CancellationToken;
pub async fn worker(token: CancellationToken) {
loop {
tokio::select! {
_ = token.cancelled() => {
// Cleanup
break;
}
_ = do_work() => {
// Continue
}
}
}
}
// Hierarchical cancellation
let parent_token = CancellationToken::new();
let child_token = parent_token.child_token();
tokio::spawn(worker(child_token));
// Cancel all
parent_token.cancel();
Best Practices
- Use semaphores for limiting concurrent operations
- Implement graceful shutdown in all long-running tasks
- Add timeouts to external operations
- Use channels for inter-task communication
- Handle cancellation properly in all tasks
- Clean up resources in Drop or explicit shutdown methods
- Use appropriate channel types for different patterns
- Implement retries for transient failures
- Use select! for coordinating multiple async operations
- Document lifetime and ownership patterns clearly
FAQ
What does the tokio-patterns skill do?
Common Tokio patterns and idioms for async programming. Use when implementing worker pools, request-response patterns, pub/sub, timeouts, retries, or graceful shutdown.
How do I install tokio-patterns?
Run npx skills add https://github.com/geoffjay/claude-plugins --skill tokio-patterns in your project. The skills CLI detects the agents you use and copies the skill into the right folder. For Claude Code you can also place the skill folder at ~/.claude/skills/tokio-patterns/ (personal) or .claude/skills/tokio-patterns/ (project).
Which AI agents can use tokio-patterns?
Any agent that supports the open Agent Skills format (a folder with a SKILL.md file), including Claude Code, Codex, Cursor, GitHub Copilot, Gemini CLI, Windsurf, Cline, OpenCode and more.
Is tokio-patterns safe to install?
No partner security audit is available yet. Skills can run commands and scripts, so read the SKILL.md and any bundled files before installing.
Who maintains tokio-patterns?
It is published by geoffjay in geoffjay/claude-plugins. It has 117 recorded installs.
More from geoffjay/claude-plugins
documentation-update
geoffjay/claude-plugins
Regenerates documentation files (agents.md, agent-skills.md, plugins.md, usage.md) from marketplace data using Jinja templates. Use when plugins are added, updated, or removed to keep documentation in sync.
git-troubleshooting
geoffjay/claude-plugins
Git troubleshooting techniques including recovering lost commits, fixing merge conflicts, resolving detached HEAD, and diagnosing repository issues. Use when user encounters git errors or needs to recover from mistakes.
git-advanced
geoffjay/claude-plugins
Advanced git operations including complex rebase strategies, interactive staging, commit surgery, and history manipulation. Use when user needs to perform complex git operations like rewriting history or advanced merging.
tokio-concurrency
geoffjay/claude-plugins
Advanced concurrency patterns for Tokio including fan-out/fan-in, pipeline processing, rate limiting, and coordinated shutdown. Use when building high-concurrency async systems.
git-repository
geoffjay/claude-plugins
Repository management strategies including branch strategies (Git Flow, GitHub Flow, trunk-based), monorepo patterns, submodules, and repository organization. Use when user needs guidance on repository structure or branching strategies.
tokio-troubleshooting
geoffjay/claude-plugins
Debugging and troubleshooting Tokio applications using tokio-console, detecting deadlocks, memory leaks, and performance issues. Use when diagnosing async runtime problems.
Popular other skills
hyperframes-cli
heygen-com/hyperframes
The HyperFrames CLI by job - scaffold, check, look at frames, edit the timeline, preview, render locally or in the cloud, publish, project history - plus reading its JSON and recovering from build or render failures.
triage
mattpocock/skills
Move issues and external PRs through a state machine of triage roles, categorise, verify, grill if needed, and write agent-ready briefs.
ask-matt
mattpocock/skills
Ask which skill or flow fits your situation. A router over the skills in this repo.
to-questionnaire
mattpocock/skills
Turn a decision you can't fully answer into a questionnaire for someone else to fill in.
loop-me
mattpocock/skills
Grill me about specs for the workflows I want to build, within this workspace.