Rustのasyncをもっと深く：select!・タイムアウト・ストリーム・Pin
async/awaitのページでは基本を学びました。async fn は遅延評価される Future を返し、.await でそれを実行し、#[tokio::main] がランタイムを起動し、tokio::join! と tokio::spawn で複数の処理を同時に動かせる、という内容でした。まだ自信がなければ、先にそちらを読んでください。このページはその続きです。
実際の非同期プログラムには、もう少し道具が必要です。また忙しい厨房を思い浮かべてみましょう。すべての料理ができるまで待つこともあります（join!）。最初に起きたことだけが大事な場合もあります。料理ができたのか、お客さんが帰ってしまったのか（select!）。5分待ってだめならあきらめることもあります（timeout）。そして、注文がひと晩じゅう1件ずつ届き続けることもあります（ストリーム）。このページではこれらをすべて扱います。さらに、いつかエラーメッセージで出会う不思議な言葉 Pin についても、やさしい言葉で説明します。
例はすべて、基本のページと同じく features = ["full"] を指定した tokio を使います。Rust Playground には tokio が最初から入っているので、「▶ 実際に試してみよう」ボタンもそのまま使えます。
たくさんの処理を同時に動かす
Section titled “たくさんの処理を同時に動かす”tokio::join! はもう知っていますね。すべての Future が終わるのを待ち、結果をタプルで返します。その兄弟の tokio::try_join! は、Result を返す Future のためのものです。すべて成功すれば値をまとめた Ok を、どれかが失敗すれば最初の Err を返します。ひとつでも失敗した時点で、ほかの Future を待つのをやめます。
use std::time::Duration;use tokio::time::sleep;
async fn load_profile(id: u32) -> Result<String, String> { sleep(Duration::from_millis(20)).await; Ok(format!("プロフィール #{id}"))}
async fn load_settings(id: u32) -> Result<String, String> { sleep(Duration::from_millis(10)).await; if id == 2 { return Err(format!("ユーザー{id}の設定がありません")); } Ok(String::from("ダークモード"))}
#[tokio::main]async fn main() { for id in [1, 2] { match tokio::try_join!(load_profile(id), load_settings(id)) { Ok((profile, settings)) => println!("ユーザー{id}: {profile}、{settings}"), Err(e) => println!("ユーザー{id}の読み込みに失敗: {e}"), } }}実行結果:
ユーザー1: プロフィール #1、ダークモードユーザー2の読み込みに失敗: ユーザー2の設定がありませんjoin! と try_join! は、Future を今のタスクの中で順番に少しずつ動かします。最初から数がわかっている少数の Future にはこれでぴったりです。でも、実行してみるまで数がわからないジョブのリストなら、それぞれを別のタスクとして起動しましょう。基本のページのように JoinHandle を Vec に入れていく方法もあります。ですが tokio にはもっと便利な道具があります。それが JoinSet です。タスクを入れておく袋のようなもので、タスクが終わった順に結果を渡してくれます。
use std::time::Duration;use tokio::task::JoinSet;use tokio::time::sleep;
async fn check_site(site: &'static str, ms: u64) -> (&'static str, u64) { sleep(Duration::from_millis(ms)).await; // ネットワーク通信のつもり (site, ms)}
#[tokio::main]async fn main() { let mut set = JoinSet::new(); for (site, ms) in [("rust-lang.org", 30), ("docs.rs", 10), ("crates.io", 20)] { set.spawn(check_site(site, ms)); }
let mut results = Vec::new(); while let Some(res) = set.join_next().await { results.push(res.unwrap()); // Err になるのはタスクがパニックしたときだけ }
results.sort(); // 終わる順番は実行のたびに変わることがある for (site, ms) in results { println!("{site} の応答時間: {ms}ms"); }}実行結果:
crates.io の応答時間: 20msdocs.rs の応答時間: 10msrust-lang.org の応答時間: 30msjoin_next() は袋が空になると None を返すので、while let ループは自然に終わります。結果は起動した順ではなく、終わった順に届きます。かかる時間が近いタスクどうしだと、その順番は実行のたびに変わることがあるので、表示する前に並べ替えています。もうひとつうれしい点があります。JoinSet をドロップすると、中に残っているタスクはすべてキャンセルされます。うっかり何かが動き続けてしまうことはありません。
select! で Future を競争させる
Section titled “select! で Future を競争させる”join! はすべてを待ちます。tokio::select! は最初の1つを待ちます。各ブランチ（branch）は、パターン、Future、コードのブロックでできていて、pattern = future => code と書きます。最初に終わった Future のコードが実行され、ほかのブランチはキャンセルされます。
Rust でキャンセルされるとは、どういうことでしょうか。負けた Future は、スコープを抜けるほかの値と同じように、ただドロップされるだけです。二度と実行されることはありません。その中のローカル変数もドロップされるので、後片付けのコードはきちんと実行されます。次の例では、小さな Drop 型を使ってそれを目で見えるようにしています。
use std::time::Duration;use tokio::time::sleep;
struct Download(&'static str);
impl Drop for Download { fn drop(&mut self) { println!("{}のダウンロードを片付けます", self.0); }}
async fn download(mirror: &'static str, ms: u64) -> &'static str { let _download = Download(mirror); sleep(Duration::from_millis(ms)).await; println!("{mirror}のダウンロードが完了しました"); mirror}
#[tokio::main]async fn main() { tokio::select! { winner = download("ヨーロッパ", 50) => println!("勝者: {winner}"), winner = download("アジア", 1000) => println!("勝者: {winner}"), } println!("main は続きます");}実行結果:
ヨーロッパのダウンロードが完了しましたヨーロッパのダウンロードを片付けますアジアのダウンロードを片付けます勝者: ヨーロッパmain は続きます何が表示されていないかに注目してください。「アジアのダウンロードが完了しました」は一度も表示されません。アジアの Future は .await のところで止められてドロップされ、その Download の値もすぐに片付けられました。非同期 Rust のキャンセルは、どこでもこの仕組みです。つまり Future をポーリングするのをやめて、ドロップする だけです。特別な「キャンセル」の呼び出しは必要ありません。
select! の定番の使い方がタイムアウトです。自分の処理とタイマーを競争させます。
tokio::select! { report = build_report() => println!("{report}"), _ = tokio::time::sleep(Duration::from_secs(5)) => println!("遅すぎるのであきらめます"),}これでもちゃんと動きますが、専用の近道 tokio::time::timeout があります。次の節で紹介します。
停止の合図が来るまで働く
Section titled “停止の合図が来るまで働く”select! が本当に活躍するのは loop の中です。次の例では、ワーカーがチャネルからジョブを受け取りながら、2本目のチャネルで「止まってください」というメッセージも待っています。oneshot チャネルはメッセージをちょうど1つだけ運ぶので、停止の合図にぴったりです。
use std::time::Duration;use tokio::sync::{mpsc, oneshot};use tokio::time::sleep;
#[tokio::main]async fn main() { let (job_tx, mut job_rx) = mpsc::channel::<String>(10); let (stop_tx, mut stop_rx) = oneshot::channel::<()>();
let worker = tokio::spawn(async move { let mut done = 0; loop { tokio::select! { Some(job) = job_rx.recv() => { println!("作業中: {job}"); done += 1; } _ = &mut stop_rx => { println!("停止の合図を受け取りました"); break; } } } done });
for job in ["写真のリサイズ", "メールの送信", "検索インデックスの更新"] { job_tx.send(job.to_string()).await.unwrap(); } sleep(Duration::from_millis(100)).await; // ワーカーが終えるまで少し待つ stop_tx.send(()).unwrap();
let done = worker.await.unwrap(); println!("ワーカーは{done}件のジョブを終えました");}実行結果:
作業中: 写真のリサイズ作業中: メールの送信作業中: 検索インデックスの更新停止の合図を受け取りましたワーカーは3件のジョブを終えました知っておくと役に立つ細かい点をいくつか挙げます。
stop_rxではなく&mut stop_rxと書いています。ループのたびにselect!は同じ受信側をもう一度使う必要があるので、1周目でムーブしてしまわずに貸しているのです。- ループのたびに
job_rx.recv()は新しい Future を作り、負けたほうはドロップされます。それでも問題ないのは、recv()がキャンセル安全（cancel-safe）だからです。ドロップしてもメッセージが失われることはありません。 - パターンにマッチしなかった場合（ここでは、送信側がすべてなくなって
recv()がNoneを返した場合）、select!はそのブランチを無効にして、残りのブランチを待ち続けます。 - 複数のブランチが同時に準備できていたら、
select!はランダムに1つを選びます。特定のブランチがループを独占しないようにするためです。上から順番に調べてほしいときは、select!の中の1行目にbiased;と書きます。
キャンセル安全性をひと言で。 負けたブランチはドロップされるので、途中までやっていた作業は捨てられます。チャネルの recv() や sleep() なら困りません。まだ何も取り出していないからです。でも、ネットワークのソケットから100バイト読む Future が、60バイト読んだところでドロップされたらどうでしょう。その60バイトは消えてしまいます。こういう Future はキャンセル安全ではありません。tokio のドキュメントには、メソッドごとにキャンセル安全かどうかが書かれています。ループの中で select! を使うときは、呼び出すメソッドを確認しましょう。あるいは、Future をループの外で一度だけ作り、その同じ Future をポーリングし続ける方法もあります（やり方は Pin の節で紹介します）。
タイムアウトとタイマー
Section titled “タイムアウトとタイマー”tokio::time::timeout(duration, future) は、どんな Future でも包めます。時間内に終われば Ok(value) が返ります。先に時間切れになると、Future はドロップされ（select! で負けたブランチと同じようにキャンセルされます）、Err(Elapsed) が返ります。
use std::time::Duration;use tokio::time::{sleep, timeout};
async fn fetch_price(server_delay_ms: u64) -> u32 { sleep(Duration::from_millis(server_delay_ms)).await; 42}
#[tokio::main]async fn main() { let limit = Duration::from_millis(200);
for (server, delay) in [("速いサーバー", 20), ("遅いサーバー", 2000)] { match timeout(limit, fetch_price(delay)).await { Ok(price) => println!("{server}: 価格は{price}です"), Err(e) => println!("{server}: {e}"), } }}実行結果:
速いサーバー: 価格は42です遅いサーバー: deadline has elapsed「deadline has elapsed」（期限が過ぎました）は、tokio の Elapsed エラーが表示する英語のメッセージです。外の世界とやりとりする処理には、必ずタイムアウトを付けましょう。HTTP 呼び出し、データベースへの問い合わせ、ソケットからの読み込みなどです。いつまでも返事をしないサーバーのせいで、プログラムが永遠に固まってしまってはいけません。
sleep と interval の違い。 sleep は1回だけ待ちます。interval はメトロノームのように一定のリズムを刻みます。ループの中で .tick().await を呼ぶと、予定どおりのタイミングで tick が届きます。最初の tick はすぐに届きます。
use std::time::Duration;use tokio::time::interval;
#[tokio::main]async fn main() { let mut heartbeat = interval(Duration::from_millis(50));
for beat in 1..=3 { heartbeat.tick().await; // 最初の tick はすぐに完了する println!("ハートビート {beat}"); }}実行結果:
ハートビート 1ハートビート 2ハートビート 3ループの中で50ms sleep するだけではだめなのでしょうか。それだと1周ごとに「処理の時間＋50ms」かかるので、予定が少しずつずれていきます。interval は各 tick の予定時刻から測るので、処理に10msかかったなら、残りの40msだけ待ちます。ハートビート、定期的なポーリング、「1分ごとに保存」のようなジョブには interval を、1回だけ間を空けたいときは sleep を使いましょう。
ストリーム：非同期のイテレータ
Section titled “ストリーム：非同期のイテレータ”イテレータは next() で値を1つずつ渡してくれます。ストリーム（stream）はその非同期版です。next().await は次の値を準備ができたときに渡してくれて、もう値がなければ None を返します。株価のティッカー、データベースから届く行、チャットサーバーからのメッセージなどを思い浮かべてください。
実は、ストリームのようなものはもう使ったことがあります。tokio のチャネルの受信側です。rx.recv().await は次のメッセージを待ち、送信側がすべてなくなると None を返します。
use std::time::Duration;use tokio::sync::mpsc;use tokio::time::sleep;
#[tokio::main]async fn main() { let (tx, mut rx) = mpsc::channel(8);
tokio::spawn(async move { for temp in [21, 22, 24, 23] { tx.send(temp).await.unwrap(); sleep(Duration::from_millis(10)).await; // 10msごとに計測 } // ここで tx がドロップされ、チャネルが閉じる });
while let Some(temp) = rx.recv().await { println!("気温: {temp}°C"); } println!("センサーが止まりました");}実行結果:
気温: 21°C気温: 22°C気温: 24°C気温: 23°Cセンサーが止まりました標準ライブラリにはまだ Stream トレイトがありません（今も設計中です）。エコシステムでは futures クレートのものが使われていて、tokio の補助クレート tokio-stream が StreamExt トレイトを追加します。next()、map()、filter()、take() など、イテレータでおなじみのメソッドが使えるようになります。プロジェクトに追加しましょう。
[dependencies]tokio = { version = "1", features = ["full"] }tokio-stream = "0.1"use tokio_stream::{self as stream, StreamExt};
#[tokio::main]async fn main() { let prices = stream::iter(vec![100, 102, 98, 105, 110]);
let mut doubled = prices.map(|p| p * 2).take(3);
while let Some(p) = doubled.next().await { println!("{p}"); }}これは 200、204、196 と表示します。next() に .await が付いていること以外は、イテレータのコードとそっくりですね。tokio のチャネルの受信側を本物のストリームにしたいときは、tokio_stream::wrappers::ReceiverStream::new(rx) で包みます（futures クレートにも同じ考え方の StreamExt があります。どちらのトレイトにも next() メソッドがあってぶつかるので、1つのファイルではどちらか一方だけを使いましょう）。
Pin をやさしい言葉で
Section titled “Pin をやさしい言葉で”遅かれ早かれ、型やエラーメッセージで Pin を目にするでしょう。ここでは難しい理論は抜きにして、考え方だけを説明します。
Rust は async fn をコンパイルするとき、それを隠れた構造体、つまりステートマシン（状態機械）に変えます。.await のあとでもまだ必要なローカル変数は、すべてその構造体のフィールドになります。ここで、あるローカル変数が別のローカル変数を借用しているとしましょう。たとえば let name = String::from("Ana"); let r = &name; の間に .await がある場合です。すると構造体は、自分自身のフィールドを指すポインタを持つことになります。この構造体をメモリ上の別の場所に動かしたら、そのポインタは古い、空っぽの場所を指してしまいます。
それを解決するのが Pin です。ピン留め（pin）された値は、「この値はもうメモリ上で動きません」 という約束です。Future をピン留めしてしまえば、安心してポーリングできます。ふだんはまったく気になりません。.await が代わりにピン留めしてくれるからです。
Pin に出会うのは、次の3つの場面です。
- 同じ Future を何度もポーリングするとき。 たとえば
select!ループの中の締め切り（deadline）です。 - Future を保存するとき。
Vecや構造体のフィールドに入れる場合で、たいていはPin<Box<dyn Future<Output = T>>>という形になります。 async fnの中で再帰するとき。
1つ目について、ピン留めせずにループの中で sleep を使い回そうとするとどうなるか見てみましょう。
let deadline = sleep(Duration::from_millis(300));
loop { tokio::select! { _ = &mut deadline => break, Some(n) = rx.recv() => println!("{n}"), }}error[E0277]: `PhantomPinned` cannot be unpinned | = note: consider using the `pin!` macro consider using `Box::pin` if you need to access the pinned value outside of the current scope = note: required for `Sleep` to implement `Unpin`エラーメッセージは「PhantomPinned はピン留めを外せない」という意味ですが、うれしいことにコンパイラが直し方まで教えてくれています。std::pin::pin!（または tokio::pin!）は、Future をその場でスタック上にピン留めします。これで、ループの毎回の周回で同じ締め切りを共有できます。
use std::pin::pin;use std::time::Duration;use tokio::sync::mpsc;use tokio::time::sleep;
#[tokio::main]async fn main() { let (tx, mut rx) = mpsc::channel(8); tokio::spawn(async move { for order in 1..=3 { tx.send(order).await.unwrap(); sleep(Duration::from_millis(10)).await; } sleep(Duration::from_secs(5)).await; // そのあとお店は静かになる });
let mut deadline = pin!(sleep(Duration::from_millis(300)));
loop { tokio::select! { _ = &mut deadline => { println!("閉店時間です"); break; } Some(order) = rx.recv() => println!("注文 #{order} を受け付けました"), } }}実行結果:
注文 #1 を受け付けました注文 #2 を受け付けました注文 #3 を受け付けました閉店時間ですもし select! の中で sleep(...) と書いていたら、注文が来るたびに新しい300msのタイマーが始まり、忙しいお店はいつまでたっても閉店できません。ループの外で締め切りを1つだけピン留めすれば、この問題は解決します。
再帰の場合、コンパイラは Future の大きさを知る必要があります。ところが自分自身を含む Future は、無限に大きくなってしまいます。async fn countdown の中で countdown(n - 1).await を呼ぶと、次のエラーになります。
error[E0733]: recursion in an async fn requires boxing | = note: a recursive `async fn` call must introduce indirection such as `Box::pin` to avoid an infinitely sized future「async fn の再帰にはボックス化が必要です」という意味です。Box::pin は内側の Future をヒープに移して、そこでピン留めします。外側の Future が持つのはポインタだけになり、その大きさは決まっています。
async fn countdown(n: u32) { if n == 0 { println!("発射！"); return; } println!("{n}..."); Box::pin(countdown(n - 1)).await;}
#[tokio::main]async fn main() { countdown(3).await;}実行結果:
3...2...1...発射！正直なところ、ほとんどの人に必要なのはこれだけです。何度もポーリングする Future には pin!、Future の保存や再帰には Box::pin です。くわしく知りたくなったら、「さらに詳しく」にある Rust Book の章が順を追って説明してくれます。
トレイトの async fn と非同期クロージャ
Section titled “トレイトの async fn と非同期クロージャ”Rust 1.75 からは、トレイトの中に直接 async fn を書けるようになりました。Rust 1.85 からは非同期クロージャ（async closure）も使えます。async |x| ... と書き、対応するトレイトとして AsyncFn、AsyncFnMut、AsyncFnOnce があります（クロージャのトレイトの非同期版です）。1つの例で両方を見てみましょう。
use std::time::Duration;use tokio::time::sleep;
trait Notifier { async fn notify(&self, user: &str) -> String;}
struct Email;
impl Notifier for Email { async fn notify(&self, user: &str) -> String { sleep(Duration::from_millis(10)).await; // メールサーバーとやりとりするつもり format!("{user}さんにメールを送りました") }}
async fn for_each_user(users: &[&str], job: impl AsyncFn(&str) -> String) { for user in users { println!("{}", job(user).await); }}
#[tokio::main]async fn main() { let email = Email; let users = ["ana", "ben"];
for_each_user(&users, async |user: &str| email.notify(user).await).await;}実行結果:
anaさんにメールを送りましたbenさんにメールを送りましたトレイトの async fn には、1つ落とし穴があります。トレイトは、返される Future が Send であることを約束していません。そして tokio::spawn には Send な Future が必要です（スレッドの間を移動するかもしれないからです）。そのため、ジェネリックな N: Notifier を通した呼び出しを spawn しようとすると失敗します。
fn send_in_background<N: Notifier + Send + Sync + 'static>(notifier: N) { tokio::spawn(async move { let msg = notifier.notify("ana").await; println!("{msg}"); });}error: future cannot be sent between threads safely |7 | let msg = notifier.notify("ana").await; | ^^^^^^^^^^^^^^^^^^^^^^ await occurs here on type `impl Future<Output = String>`, which is not `Send` |help: `Send` can be made part of the associated future's guarantees for all implementations of `Notifier::notify` |2 - async fn notify(&self, user: &str) -> String;2 + fn notify(&self, user: &str) -> impl std::future::Future<Output = String> + Send;「Future をスレッド間で安全に送れません」というエラーです。コンパイラの提案が、いつもの直し方です。トレイトのメソッドを fn ... -> impl Future<Output = String> + Send と書きます。実装する側は、これまでどおり普通の async fn で書けます。また、async fn を持つトレイトは、今のところ dyn Notifier というトレイトオブジェクトとしては使えません。それが必要なら、人気の async-trait クレートが Future をボックス化してくれます。
よくある間違い
Section titled “よくある間違い”ランタイムをブロックしてしまう。 非同期タスクの中で std::thread::sleep や重い計算、大きな std::fs の読み込みをすると、ワーカースレッド全体が固まり、そのスレッド上のほかのタスクもすべて待たされます。基本のページでは sleep の例を見ました。CPU を多く使う処理やブロックする処理は、tokio::task::spawn_blocking に渡しましょう。そういう処理のために用意された、別のスレッドプールで実行してくれます。
fn count_primes(limit: u64) -> usize { (2..limit) .filter(|&n| (2..).take_while(|d| d * d <= n).all(|d| n % d != 0)) .count()}
#[tokio::main]async fn main() { let handle = tokio::task::spawn_blocking(|| count_primes(50_000));
println!("数えているあいだも、リクエストに応答しています...");
let primes = handle.await.unwrap(); println!("50000未満の素数は{primes}個あります");}実行結果:
数えているあいだも、リクエストに応答しています...50000未満の素数は5133個ありますstd::sync::Mutex のガードを .await をまたいで持ち続ける。 一見、何の問題もなさそうです。
let handle = tokio::spawn(async move { let mut count = visits.lock().unwrap(); *count += 1; save_to_disk(*count).await;});error: future cannot be sent between threads safely | = help: within `{async block@src/main.rs:13:31: 13:41}`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, u32>`note: future is not `Send` as this value is used across an await |14 | let mut count = visits.lock().unwrap(); | --------- has type `std::sync::MutexGuard<'_, u32>` which is not `Send`15 | *count += 1;16 | save_to_disk(*count).await; | ^^^^^ await occurs here, with `mut count` maybe used laterこれはコンパイラがあなたを守ってくれているのです。このタスクが .await で眠っているあいだもロックを持ったままになり、そのロックを欲しがるほかのタスクは、自分のスレッドごとブロックしてしまいます。直し方は、.await の前にガードをドロップすることです。ロックする部分を独立したブロックに入れましょう。
let count = { let mut count = visits.lock().unwrap(); *count += 1; *count}; // ここでガードがドロップされるsave_to_disk(count).await;どうしても待っているあいだもロックを持ち続けたいなら、tokio::sync::Mutex を使います。その lock().await は、スレッドをブロックせずに待ってくれます（ロックについてくわしくは共有状態を見てください）。
.await を書き忘れる。 Future は遅延評価されるので、.await なしの save_file(); はまったく何もしません。コンパイラは「futures do nothing unless you .await or poll them」（Future は .await するかポーリングしない限り何もしない）と警告してくれます。くわしくは async/await を見てください。この警告は無視しないようにしましょう。
タスクを spawn したまま待たない。 main が終わるとランタイムが停止し、終わっていないタスクはすべてドロップされます。
use std::time::Duration;use tokio::time::sleep;
#[tokio::main]async fn main() { tokio::spawn(async { sleep(Duration::from_millis(100)).await; println!("レポートを保存しました"); // 表示されない！ });
println!("main が終わりました");}実行結果:
main が終わりましたレポートは保存されません。大事な処理なら、main が終わる前に JoinHandle を取っておいて .await しましょう（または、タスクを JoinSet に入れて、全部取り出しましょう）。
1. 締め切り付きの価格チェック
Section titled “1. 締め切り付きの価格チェック”4つのお店が、価格の問い合わせにそれぞれ20、40、300、60ミリ秒後に答えます。JoinSet を使って4店に同時に問い合わせますが、各店に与える時間は150msだけです。届いた価格を（店名順に並べて）表示し、そのあとタイムアウトした店の数を表示してください。ヒント： timeout(limit, ask_shop(name, delay)) を袋に spawn します。終わったタスクはそれぞれ Ok か Err(Elapsed) を返すので、店名をその結果と一緒に返しましょう。
解答を見る
use std::time::Duration;use tokio::task::JoinSet;use tokio::time::{sleep, timeout};
async fn ask_shop(delay_ms: u64) -> u32 { sleep(Duration::from_millis(delay_ms)).await; 100 + delay_ms as u32 // 適当な価格}
#[tokio::main]async fn main() { let shops = [("Corner Shop", 20), ("MegaMart", 40), ("SlowCo", 300), ("Bits & Bobs", 60)]; let limit = Duration::from_millis(150);
let mut set = JoinSet::new(); for (name, delay) in shops { set.spawn(async move { (name, timeout(limit, ask_shop(delay)).await) }); }
let mut prices = Vec::new(); let mut timed_out = 0; while let Some(res) = set.join_next().await { match res.unwrap() { (name, Ok(price)) => prices.push((name, price)), (_, Err(_)) => timed_out += 1, } }
prices.sort(); for (name, price) in prices { println!("{name}: {price}"); } println!("{timed_out}店がタイムアウトしました");}実行結果:
Bits & Bobs: 160Corner Shop: 120MegaMart: 1401店がタイムアウトしましたすべてのお店を同時に待つので、SlowCo がどんなに遅くても、チェック全体は150msほどで終わります。SlowCo のタイムアウトが来るとその Future はドロップされるので、裏で動き続けることもありません。店名は英語のままにしています。並べ替えはアルファベット順なので、そのほうが結果がわかりやすいからです。
2. 非同期の再帰でフォルダのサイズを求める
Section titled “2. 非同期の再帰でフォルダのサイズを求める”Folder は、名前、ファイルサイズ（KB）のリスト、サブフォルダのリストを持っています。自分のファイルとすべてのサブフォルダのサイズを合計する async fn total_size(folder: &Folder) -> u64 を書いてください。サブフォルダごとに自分自身を呼び出します。各フォルダでは、まず tokio::task::yield_now().await で短い非同期の「ディスク読み込み」をするふりをしましょう。ヒント： 再帰呼び出しは Box::pin(...) で包む必要があります。そうしないとエラー E0733 になります。
解答を見る
struct Folder { name: &'static str, files_kb: Vec<u64>, subfolders: Vec<Folder>,}
async fn total_size(folder: &Folder) -> u64 { tokio::task::yield_now().await; // ディスクを読むふり
let mut total: u64 = folder.files_kb.iter().sum(); for sub in &folder.subfolders { total += Box::pin(total_size(sub)).await; } println!("{}: {} KB", folder.name, total); total}
#[tokio::main]async fn main() { let home = Folder { name: "home", files_kb: vec![4], subfolders: vec![ Folder { name: "photos", files_kb: vec![2048, 1024], subfolders: vec![] }, Folder { name: "code", files_kb: vec![12, 8], subfolders: vec![Folder { name: "target", files_kb: vec![500], subfolders: vec![] }], }, ], };
let total = total_size(&home).await; println!("合計: {total} KB");}実行結果:
photos: 3072 KBtarget: 500 KBcode: 520 KBhome: 3596 KB合計: 3596 KB各フォルダは、すべてのサブフォルダのあとに表示されます。先にサブフォルダのサイズがわかるのを待たなければならないからです。Box::pin は入れ子になった呼び出しの Future をヒープに置くので、外側の Future の大きさは一定のままです。
3. キューを空にしてから止まる
Section titled “3. キューを空にしてから止まる”停止の例を書き換えて、停止の合図を送る前の sleep(100ms) をなくしてください。3つのジョブと停止の合図をすぐに送りますが、ワーカーは止まる前に、キューにあるすべてのジョブを必ず終えるようにします。ヒント： 何もしなければ、両方の準備ができているとき select! はランダムにブランチを選びます。biased; を追加して、ジョブのブランチを先頭に置きましょう。
解答を見る
use tokio::sync::{mpsc, oneshot};
#[tokio::main]async fn main() { let (job_tx, mut job_rx) = mpsc::channel::<String>(10); let (stop_tx, mut stop_rx) = oneshot::channel::<()>();
for job in ["写真のリサイズ", "メールの送信", "検索インデックスの更新"] { job_tx.send(job.to_string()).await.unwrap(); } stop_tx.send(()).unwrap();
let worker = tokio::spawn(async move { let mut done = 0; loop { tokio::select! { biased; Some(job) = job_rx.recv() => { println!("作業中: {job}"); done += 1; } _ = &mut stop_rx => { println!("キューが空になったので停止します"); break; } } } done });
let done = worker.await.unwrap(); println!("ワーカーは{done}件のジョブを終えました");}実行結果:
作業中: 写真のリサイズ作業中: メールの送信作業中: 検索インデックスの更新キューが空になったので停止しますワーカーは3件のジョブを終えましたbiased; があると、select! はいつもジョブのブランチを先に試します。停止の合図を見るのは、recv() ですぐに受け取れるものがなくなってからです。これがないと、ワーカーは1つか2つのジョブを終えたところで止まってしまうかもしれません。たまにしか起きない、やっかいなバグになります。
join!はすべての Future を待ち、try_join!は最初のErrで止まります。JoinSetは spawn したたくさんのタスクを動かし、終わった順に結果を返します（順番が大事なら並べ替えましょう）。select!は最初に終わる Future を待ちます。ほかの Future はドロップされます。これが非同期 Rust のキャンセルの仕組みです。ループの中では、キャンセル安全な Future を使い、順番が大事ならbiased;を使いましょう。timeoutは遅い Future をあきらめます。intervalは一定の間隔で tick を刻み、sleepは1回だけ待ちます。- ストリームは非同期のイテレータです。チャネルの受信側はストリームのように使えます。
tokio-streamを使うとmapやtakeなどが使えます。 Pinは、Future がメモリ上で動かないことを約束します。1つの Future を何度もポーリングするならpin!、Future を保存するときや非同期の再帰にはBox::pinを使います。- ランタイムをブロックしない（
spawn_blockingを使う）、stdのロックを.awaitをまたいで持たない、spawn したタスクは待つ。この3つを覚えておきましょう。
理解度チェック
Section titled “理解度チェック”理解度チェック
1. tokio::select! で、最初に終わらなかったブランチはどうなりますか？
2. 時間切れになったとき、tokio::time::timeout は何を返しますか？
3. async fn が自分自身を呼び出すとき、なぜ Box::pin が必要なのですか？
4. tokio のプログラムの中で、CPU を多く使う2秒間の処理をしなければなりません。いちばん良い道具はどれですか？
スコア: 0 / 4
さらに詳しく
Section titled “さらに詳しく”- Select — Tokio チュートリアル（英語）：
select!をくわしく解説しています。ブランチの動き方、キャンセルの意味、ループの中での使い方がわかります。 - Streams — Tokio チュートリアル（英語）：
tokio-streamを使ったストリームの使い方、変換のしかた、自作のしかたを説明しています。 - A Closer Look at the Traits for Async — The Rust Book（英語）：
Future、Pin、Unpin、Streamを、一歩ずつやさしく見ていく章です。日本語版の Rust Book にはまだ async の章がありません。 - std::pin — Rust 標準ライブラリ（英語）：ピン留め、
Pin、Unpin、pin!マクロについての公式の説明です。 - docs.rs の tokio::time（英語）：
sleep、timeout、intervalと、tick に間に合わなかったときの扱いについて説明しています。