Rust Tokioタスクバジェット：協調スケジューリングとスターベーション防止の5つのコア戦略
编程语言
Tokioランタイムが「ストライキ」を起こしたら
本番サービスのP99レイテンシが突然数秒に急上昇、しかしCPU使用率は30%のまま——これは負荷過多ではなく、タスクスターベーションです。CPU集約型の非同期タスクがワーカースレッドを独占し、他のタスクがスケジューリングを待って行列に並んでいます。Tokioの協調スケジューリングメカニズムがこれを防ぐはずなのに、コードにyieldポイントがないため機能していません。
Tokioのタスクバジェット（Task Budget）はランタイムスケジューリングの中核防衛線です：各タスクは一定量の作業を実行した後、スレッドを明け渡し、他のタスクに実行機会を与える必要があります。このメカニズムを理解し正しく適用することは、「動く非同期コードを書く」から「プロダクション級非同期システムを書く」への重要な飛躍です。
コア概念一覧
| 概念 | 説明 | 典型的なユースケース |
|---|---|---|
| Task Budget | タスクが連続実行できるIO操作の割り当て量 | 単一タスクのスレッド独占防止 |
| Cooperative Scheduling | タスクが自発的に実行権を譲渡するスケジューリングモデル | すべてのTokio非同期コード |
yield_now |
明示的にスレッドを譲渡、協調ポイントを挿入 | CPU集約ループ、長時間計算 |
spawn_blocking |
ブロッキングタスクを専用スレッドプールに移動 | ファイルIO、暗号計算、同期API |
| IOバジェット | Tokioの連続IO操作に対する割り当て制限 | IO集約タスクが他タスクをスターベーションさせるのを防止 |
| ワークスティーリング | アイドルスレッドがビジースレッドからタスクを盗む | マルチスレッドランタイム負荷分散 |
| タスク優先度 | スケジューリング戦略でタスク実行順序を制御 | クリティカルパス優先、レイテンシ敏感タスク |
| スターベーション検出 | ランタイムが長時間スケジューリングされていないタスクを検出 | 監視アラート、パフォーマンス診断 |
5つのスケジューリング課題
1. CPU集約タスクがランタイムをブロック
非同期関数内でCPU集約計算（JSONパース、暗号化、ソートなど）を実行し、yieldポイントを挿入しないとワーカースレッドを独占し、同じスレッドの他タスクがすべてスターベーションします。
2. タスクスターベーション検出の困難さ
スターベーションは「サイレント障害」です——パニックもエラーログもなく、レイテンシ曲線の緩やかな上昇だけが現れます。複雑なシステムでどのタスクがスターベーションを引き起こしているか特定するには、逐次調査が必要です。
3. IOバジェットの枯渇
Tokioは各タスクにIO操作バジェット（デフォルト128回）を割り当て、バジェットが尽きるとタスクは強制的に譲渡されます。このメカニズムを理解していないと、パフォーマンスの揺らぎと誤認する可能性があります。
4. 優先度逆転
低優先度タスクがリソース（Mutexなど）を保持し、高優先度タスクがそのリソースを待ち、中優先度タスクが実行し続ける——古典的な優先度逆転は非同期ランタイムにも存在します。
5. ランタイムチューニングパラメータ
worker_threads、max_blocking_threads、thread_stack_sizeなどのパラメータをどう設定するか？デフォルト値はプロダクション環境では最適ではないことが多いです。
5つのコア戦略
戦略1：yield_now協調ポイント挿入
CPU集約ループにyield_now()を挿入し、自発的にスレッドを他のタスクに譲渡します。
use tokio::task::yield_now;
async fn cpu_intensive_with_yields(data: &[u64]) -> u64 {
let mut sum = 0u64;
for (i, &val) in data.iter().enumerate() {
sum = sum.wrapping_add(val);
if i % 1024 == 0 {
yield_now().await;
}
}
sum
}
#[tokio::main]
async fn main() {
let data: Vec<u64> = (0..1_000_000).collect();
let compute = tokio::spawn(cpu_intensive_with_yields(&data));
let heartbeat = tokio::spawn(async {
let mut tick = 0u32;
loop {
tokio::time::sleep(std::time::Duration::from_millis(10)).await;
tick += 1;
println!("heartbeat tick: {}", tick);
if tick >= 20 {
break;
}
}
});
let result = compute.await.unwrap();
heartbeat.await.unwrap();
println!("sum: {}", result);
}
重要ポイント：
- 1024回反復ごとにyieldし、スループットと公平性のバランスを取る
yield_now().awaitは現在のタスクをスケジューリングキューの末尾に移動- ハートビートタスクは計算タスクにスターベーションされない
戦略2：spawn_blockingによるCPU集約タスク分離
CPU集約またはブロッキング操作を専用ブロッキングスレッドプールに移動し、ワーカースレッドを完全に解放します。
use tokio::task::spawn_blocking;
use std::sync::Arc;
fn heavy_computation(data: Vec<u64>) -> u64 {
data.iter().fold(0u64, |acc, &v| acc.wrapping_add(v))
}
async fn process_with_blocking_pool(data: Arc<Vec<u64>>) -> u64 {
let data_clone = Arc::clone(&data);
spawn_blocking(move || heavy_computation((*data_clone).clone())).await.unwrap()
}
#[tokio::main(worker_threads = 4)]
async fn main() {
let data = Arc::new((0..10_000_000u64).collect::<Vec<_>>());
let mut handles = vec![];
for i in 0..8 {
let data = Arc::clone(&data);
handles.push(tokio::spawn(async move {
let result = process_with_blocking_pool(data).await;
println!("task-{} result: {}", i, result);
}));
}
let heartbeat = tokio::spawn(async {
for tick in 1..=10 {
tokio::time::sleep(std::time::Duration::from_millis(50)).await;
println!("heartbeat: {}", tick);
}
});
for h in handles {
h.await.unwrap();
}
heartbeat.await.unwrap();
}
重要ポイント：
spawn_blockingは独立したスレッドプール（デフォルト512スレッド）を使用、ワーカースレッドを占有しない- 暗号化、圧縮、同期データベースドライバなどのブロッキング操作に適している
Arcでデータを共有し、コピーオーバーヘッドを回避
戦略3：タスク優先度とスケジューリング制御
tokio::select!のbiasedモードとタスク分解で優先度スケジューリングを実装します。
use tokio::sync::mpsc;
use tokio::select;
#[derive(Debug)]
enum PriorityTask {
Critical { payload: String },
Normal { payload: String },
Low { payload: String },
}
async fn priority_scheduler(
mut critical_rx: mpsc::Receiver<PriorityTask>,
mut normal_rx: mpsc::Receiver<PriorityTask>,
mut low_rx: mpsc::Receiver<PriorityTask>,
) {
loop {
select! {
biased;
Some(task) = critical_rx.recv() => {
println!("CRITICAL: {:?}", task);
}
Some(task) = normal_rx.recv() => {
println!("NORMAL: {:?}", task);
}
Some(task) = low_rx.recv() => {
println!("LOW: {:?}", task);
}
else => {
println!("all channels closed");
break;
}
}
}
}
#[tokio::main]
async fn main() {
let (crit_tx, crit_rx) = mpsc::channel::<PriorityTask>(32);
let (norm_tx, norm_rx) = mpsc::channel::<PriorityTask>(64);
let (low_tx, low_rx) = mpsc::channel::<PriorityTask>(128);
let scheduler = tokio::spawn(priority_scheduler(crit_rx, norm_rx, low_rx));
low_tx.send(PriorityTask::Low { payload: "batch-job-1".into() }).await.unwrap();
norm_tx.send(PriorityTask::Normal { payload: "api-request-1".into() }).await.unwrap();
crit_tx.send(PriorityTask::Critical { payload: "health-check-fail".into() }).await.unwrap();
drop(crit_tx);
drop(norm_tx);
drop(low_tx);
scheduler.await.unwrap();
}
重要ポイント：
biasedはselect!がブランチ順に優先的にチェックすることを保証、高優先度チャネルが先に処理される- 異なる優先度は独立したチャネルを使用、低優先度メッセージが高優先度をブロックしない
- クリティカルパスタスクはより小さいチャネル容量を使用、キューイングレイテンシを削減
戦略4：IOバジェット管理とスターベーション防止
Tokioの協調IOバジェットメカニズムを理解し、IO操作密度を適切に制御します。
use tokio::net::TcpStream;
use tokio::io::AsyncReadExt;
use tokio::task::yield_now;
const IO_BUDGET_LIMIT: usize = 64;
async fn read_with_budget_control(
stream: &mut TcpStream,
buf: &mut [u8],
) -> std::io::Result<usize> {
let mut total_read = 0;
let mut io_count = 0;
loop {
match stream.read(buf).await {
Ok(0) => break,
Ok(n) => {
total_read += n;
io_count += 1;
if io_count >= IO_BUDGET_LIMIT {
yield_now().await;
io_count = 0;
}
if total_read >= buf.len() {
break;
}
}
Err(e) => return Err(e),
}
}
Ok(total_read)
}
async fn starvation_resilient_service() {
let mut handles = vec![];
for i in 0..10 {
handles.push(tokio::spawn(async move {
let mut counter = 0u64;
loop {
tokio::time::sleep(std::time::Duration::from_millis(1)).await;
counter += 1;
if counter % 1000 == 0 {
println!("task-{} alive, counter: {}", i, counter);
}
if counter >= 5000 {
break;
}
}
}));
}
for h in handles {
h.await.unwrap();
}
}
#[tokio::main]
async fn main() {
starvation_resilient_service().await;
}
重要ポイント：
- TokioのデフォルトIOバジェットは128操作、超過するとタスクは自動的に譲渡
- IO密度を手動制御（例：64操作後にyield）で、よりきめ細かい公平性を保証
- 各非同期タスクが定期的にyieldまたはawaitすることは、スターベーション防止の基本保証
戦略5：ランタイム監視とメトリクス収集
ランタイムメトリクス収集システムを構築し、タスクスターベーションとスケジューリング異常をリアルタイムで検出します。
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use tokio::time::{interval, Duration};
#[derive(Debug)]
struct RuntimeMetrics {
tasks_spawned: AtomicU64,
tasks_completed: AtomicU64,
yields: AtomicU64,
blocking_spawns: AtomicU64,
}
impl RuntimeMetrics {
fn new() -> Self {
Self {
tasks_spawned: AtomicU64::new(0),
tasks_completed: AtomicU64::new(0),
yields: AtomicU64::new(0),
blocking_spawns: AtomicU64::new(0),
}
}
fn report(&self) {
let spawned = self.tasks_spawned.load(Ordering::Relaxed);
let completed = self.tasks_completed.load(Ordering::Relaxed);
let yields = self.yields.load(Ordering::Relaxed);
let blocking = self.blocking_spawns.load(Ordering::Relaxed);
let in_flight = spawned - completed;
println!(
"metrics | in_flight: {} | yields: {} | blocking: {}",
in_flight, yields, blocking
);
}
}
async fn monitored_task(id: u32, metrics: Arc<RuntimeMetrics>) {
metrics.tasks_spawned.fetch_add(1, Ordering::Relaxed);
for i in 0..100 {
tokio::task::yield_now().await;
metrics.yields.fetch_add(1, Ordering::Relaxed);
if i % 25 == 0 {
tokio::time::sleep(Duration::from_millis(1)).await;
}
}
metrics.tasks_completed.fetch_add(1, Ordering::Relaxed);
println!("task-{} completed", id);
}
#[tokio::main]
async fn main() {
let metrics = Arc::new(RuntimeMetrics::new());
let metrics_clone = Arc::clone(&metrics);
let monitor = tokio::spawn(async move {
let mut ticker = interval(Duration::from_secs(1));
loop {
ticker.tick().await;
metrics_clone.report();
}
});
let mut handles = vec![];
for id in 0..20 {
let m = Arc::clone(&metrics);
handles.push(tokio::spawn(monitored_task(id, m)));
}
for h in handles {
h.await.unwrap();
}
monitor.abort();
metrics.report();
}
重要ポイント：
AtomicU64でロックフリーなメトリクス収集を実現、スケジューリングパフォーマンスに影響なし- 継続的に増加する
in_flight（spawn済み未完了）はスターベーションの早期シグナル - yieldカウントが低い場合、タスクに協調ポイントが不足している可能性
- 定期レポートでランタイムヘルススナップショットを提供
5つのよくある落とし穴
落とし穴1：ループ内でyieldを忘れる
// ❌ 誤り：CPU集約ループにyieldポイントがなく、ワーカースレッドを独占
async fn bad_loop(data: &[u64]) -> u64 {
data.iter().fold(0u64, |acc, &v| acc.wrapping_add(v))
}
// ✅ 正しい：定期的にyieldしてスレッドを解放
async fn good_loop(data: &[u64]) -> u64 {
let mut sum = 0u64;
for (i, &v) in data.iter().enumerate() {
sum = sum.wrapping_add(v);
if i % 4096 == 0 {
tokio::task::yield_now().await;
}
}
sum
}
落とし穴2：非同期コンテキストでブロッキングAPIを呼び出す
// ❌ 誤り：std::thread::sleepがワーカースレッド全体をブロック
async fn bad_blocking() {
std::thread::sleep(std::time::Duration::from_secs(1));
}
// ✅ 正しい：tokio非同期sleepまたはspawn_blockingを使用
async fn good_async() {
tokio::time::sleep(std::time::Duration::from_secs(1)).await;
}
落とし穴3：軽量操作にspawn_blockingを使用する
// ❌ 誤り：軽量計算にスレッド切り替えオーバーヘッドは不要
async fn bad_spawn_blocking() -> u64 {
tokio::task::spawn_blocking(|| 1 + 1).await.unwrap()
}
// ✅ 正しい：本当にブロッキングまたはCPU集約な場合のみ使用
async fn good_spawn_blocking() -> String {
tokio::task::spawn_blocking(|| {
let data = std::fs::read_to_string("/large/file.txt").unwrap();
data
}).await.unwrap()
}
落とし穴4：JoinErrorとタスクpanicを無視する
// ❌ 誤り：unwrapがパニックする可能性があり、キャンセル情報も失われる
async fn bad_join(handle: tokio::task::JoinHandle<u64>) -> u64 {
handle.await.unwrap()
}
// ✅ 正しい：JoinErrorを処理し、panicとキャンセルを区別
async fn good_join(handle: tokio::task::JoinHandle<u64>) -> Option<u64> {
match handle.await {
Ok(val) => Some(val),
Err(e) => {
if e.is_panic() {
eprintln!("task panicked: {:?}", e);
} else if e.is_cancelled() {
eprintln!("task cancelled");
}
None
}
}
}
落とし穴5：worker_threadsの過剰チューニング
// ❌ 誤り：盲目的に大量のワーカースレッドを設定
#[tokio::main(worker_threads = 64)]
async fn bad_runtime() {}
// ✅ 正しい：CPUコア数とタスクタイプに基づいて設定
#[tokio::main(worker_threads = 4)]
async fn good_runtime() {}
// IO集約：worker_threads = CPUコア数
// CPU集約：spawn_blocking + worker_threads = CPUコア数
エラートラブルシューティング早見表
| エラー現象 | 可能な原因 | 解決策 |
|---|---|---|
| P99レイテンシ急上昇だがCPU低 | タスクスターベーション、特定タスクがスレッド独占 | CPU集約ループを確認、yield_now()を追加 |
task hungログ |
タスクがバジェットを超過して譲渡なし | 連続IO操作を減らし、yieldポイントを挿入 |
JoinError::Panic |
spawnされたタスクがパニック | unwrap/expectを確認、?やmatchに変更 |
| ブロッキングスレッドプール枯渇 | spawn_blockingが多すぎる |
max_blocking_threadsを増やすかブロッキング呼び出しを削減 |
| デッドロック：タスクが永久にハング | select!のブランチが永遠にレディにならない |
タイムアウトブランチを追加するか全チャネルがアクティブであることを確認 |
| メモリが継続増加 | タスクリーク（abortされていないJoinHandle） | CancellationTokenまたはスコープ付きタスクライフサイクル管理を使用 |
| ランタイムがフリーズ無応答 | 同期コードがワーカースレッドをブロック | 非同期APIに置き換えるかspawn_blockingを使用 |
| IOスループット急減 | IOバジェット枯渇によるスケジューリング遅延 | IO密度を手動制御、定期的にyield |
Cannot start a runtime |
ネストされた#[tokio::main] |
既存ランタイム内でspawnするにはHandle::current()を使用 |
| 優先度逆転 | 低優先度がロックを保持、高優先度が待機 | ロックスコープを縮小、チャネル通信に変更 |
高度な最適化のヒント
1. 適応型yield頻度
システム負荷に基づいてyield頻度を動的に調整：低負荷時はyieldを減らしてスループットを向上、高負荷時はyieldを増やして公平性を保証。
use tokio::task::yield_now;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
struct AdaptiveYield {
interval: std::sync::atomic::AtomicUsize,
counter: std::sync::atomic::AtomicUsize,
}
impl AdaptiveYield {
fn new(base_interval: usize) -> Self {
Self {
interval: std::sync::atomic::AtomicUsize::new(base_interval),
counter: std::sync::atomic::AtomicUsize::new(0),
}
}
async fn maybe_yield(&self) {
let count = self.counter.fetch_add(1, Ordering::Relaxed) + 1;
let interval = self.interval.load(Ordering::Relaxed);
if count >= interval {
self.counter.store(0, Ordering::Relaxed);
yield_now().await;
}
}
}
2. ランタイム階層アーキテクチャ
IOタスクとCPUタスクに異なるランタイム設定を使用し、相互干渾を防止します。
use tokio::runtime::{Builder, Handle};
fn create_io_runtime() -> tokio::runtime::Runtime {
Builder::new_multi_thread()
.worker_threads(4)
.thread_name("io-worker")
.enable_all()
.build()
.unwrap()
}
fn create_compute_runtime() -> tokio::runtime::Runtime {
Builder::new_multi_thread()
.worker_threads(2)
.max_blocking_threads(64)
.thread_name("compute-worker")
.enable_all()
.build()
.unwrap()
}
3. タスクタイムアウトとサーキットブレーカー
各タスクにタイムアウトを設定し、単一タスクの異常によるシステム全体のカスケードレイテンシを防止します。
use tokio::time::{timeout, Duration};
async fn task_with_timeout(task_id: u32) -> Option<String> {
let result = timeout(Duration::from_secs(5), async {
tokio::time::sleep(Duration::from_secs(2)).await;
format!("task-{} done", task_id)
}).await;
match result {
Ok(val) => Some(val),
Err(_) => {
eprintln!("task-{} timeout", task_id);
None
}
}
}
4. カスタム協調IOバジェット
カスタムCooperativeReaderアダプタでIO操作粒度を制御します。
use tokio::io::{AsyncRead, AsyncReadExt};
use tokio::task::yield_now;
use std::pin::Pin;
use std::task::{Context, Poll};
struct CooperativeReader<R> {
inner: R,
budget: usize,
consumed: usize,
}
impl<R: AsyncRead + Unpin> AsyncRead for CooperativeReader<R> {
fn poll_read(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
buf: &mut tokio::io::ReadBuf<'_>,
) -> Poll<std::io::Result<()>> {
if self.consumed >= self.budget {
self.consumed = 0;
cx.waker().wake_by_ref();
return Poll::Pending;
}
let result = Pin::new(&mut self.inner).poll_read(cx, buf);
if result.is_ready() {
self.consumed += 1;
}
result
}
}
非同期ランタイム比較
| 特徴 | Tokio | async-std | smol | glommio |
|---|---|---|---|---|
| スケジューリングモデル | 協調型+ワークスティーリング | 協調型+ワークスティーリング | 協調型+ワークスティーリング | 協調型+スレッドローカル |
| IOバジェット | 128回/タスク | 明示的バジェットなし | 明示的バジェットなし | 明示的バジェットなし |
| ブロッキング分離 | spawn_blocking |
spawn_blocking |
spawn_blocking |
spawn_local |
| マルチスレッド | デフォルトマルチスレッド | デフォルトマルチスレッド | オプション | 主にシングルスレッド |
| エコシステム成熟度 | 最高 | 高 | 中 | 中 |
| IOドライバ | epoll/io_uring | epoll | epoll | io_uring |
| ユースケース | 汎用プロダクションシステム | 汎用システム | ライトウェイトアプリ | 高性能IO集約 |
| スターベーション防止 | IOバジェット+yield | ユーザー依存yield | ユーザー依存yield | スレッドローカル分離 |
まとめと展望
Tokioタスクバジェットのコア哲学は協調でありプリエンプションではない：各タスクが自発的に実行権を譲渡して初めて、ランタイムは公平なスケジューリングを保証できます。
yield_nowで協調ポイントを挿入、spawn_blockingでブロッキング操作を分離、biased selectで優先度を制御、IOバジェット管理でスターベーションを防止、ランタイム監視でメトリクスを収集——この5つのコア戦略を習得すれば、非同期システムはタスクスターベーションに悩まされることはありません。io_uringの深い統合と構造化並行性の導入により、Tokioのスケジューリングモデルはさらにスマートになりますが、協調スケジューリングのコア原則は変わりません——制御を譲渡してこそ、公平性を得られる。
おすすめツール
- JSONフォーマッター — TokioタスクメッセージのJSONデータをフォーマット
- ハッシュ計算 — メッセージダイジェストを計算、タスク重複排除と検証に使用
- cURL to Code — HTTPリクエストをRust reqwestコードに変換
ブラウザローカルツールを無料で試す →
#Tokio任务预算#Rust并发控制#Cooperative IO#任务调度#2026#编程语言