Zenn
🦁

Rust Async完全攻略：TokioとFuturesで高性能を引き出す10のTips

に公開
Java
Python
Rust
async
tech

Rust Async エコシステム（Tokio/Futures）の10の実践テクニック

Rust Async エコシステム（Tokio/Futures）のコアデザインはゼロコスト抽象化＋メモリ安全性にありますが、高水準な開発ではスケジューリング、メモリ、並行処理の面で隠れた落とし穴が発生しやすいです。以下の10のテクニックは、基盤となるロジックを理解し、高性能な Async コードを記述するのに役立ちます。

💡 テクニック1：Pin の本质を理解する－「固定」ではなく「約束」である

なぜこのデザインが必要か？

Async Future には自己参照（例：async fn が &self をキャプチャする場合）が含まれる可能性があります。このような Future を移動するとポインタが無効になります。Pin は物理的にメモリを「固定」するのではなく、Pin<P> 型が**「Unpin トレイトが有効になるまで、この値は移動しない」という約束**を行うものです。これは Rust における「async 安全性」と「メモリ柔軟性」のトレードオフです。

use std::pin::Pin;
use std::task::{Context, Poll};

// 自己参照型 Future の例（実開発では async fn により自動生成）
struct SelfRefFuture {
    data: String,
    ptr: *const String, // 自身の `data` フィールドを指すポインタ
}

impl Future for SelfRefFuture {
    type Output = ();
    fn poll(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Self::Output> {
        // 安全：Pin により `self` の移動が保証されるため、`ptr` は有効なまま
        let this = self.get_mut();
        unsafe { println!("{}", &*this.ptr) };
        Poll::Ready(())
    }
}

⚠️ 落とし穴回避: Unpin を手動で実装する際は、型に自己参照が存在しないことを確認してください。そうでない場合、Pin の安全性約束が破られます。

💡 テクニック2：「Async トラップ」を避ける－Sync 関数内で .await を呼び出さない

なぜこのデザインが必要か？

Rust Async のスケジューリングは協調型プリエンプションに依存しています。つまり、.await がランタイムがタスクを切り替える唯一の機会です。Sync 関数（async 修飾子なし）内で Async タスクを強制的にブロック（例：block_on 使用）すると、Tokio ワーカースレッドが占有され、他のタスクにスターベーション（飢餓） が発生します。これは、Sync 関数に「プリエンプションポイント（中断ポイント）」がないため、ランタイムが制御権を取得できないためです。

// 誤った例：Sync 関数内で Async タスクをブロック
fn sync_work() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    // 危険：タスク完了までワーカースレッドを占有し、他の Async タスクをブロック
    rt.block_on(async { fetch_data().await }); 
}

// 正しい解決策：Sync ブロッキングロジックには `spawn_blocking` を使用
async fn async_work() {
    // Tokio は Sync タスクを専用のブロッキングスレッドプールに移動し、Async スケジューリングへの干渉を回避
    tokio::task::spawn_blocking(|| sync_io_operation()).await.unwrap();
}

🔥 キーポイント: Async は「スケジューリング」を担当し、Sync は「純粋な計算／ブロッキング IO」を担当します。spawn_blocking を使用してこれらの境界を明確に分離しましょう。

💡 テクニック3：select! を JoinSet に置き換える－バッチタスク管理の最適解

なぜこのデザインが必要か？

select! は「少数のタスクを監視する」場面に適していますが、N個のタスクをバッチ処理する場合、「タスクハンドルの手動管理」が煩雑になります。Tokio 1.21 以降で導入された JoinSet は、本质的にタスクコレクションのための async キューです。自動的な結果収集、動的なタスク追加、バッチキャンセルをサポートし、内部では Sender/Receiver により効率的なスケジューリングを実現します。

use tokio::task::JoinSet;

async fn batch_fetch(urls: Vec<&str>) -> Vec<String> {
    let mut set = JoinSet::new();
    // 1. タスクをバッチで送信
    for url in urls {
        set.spawn(fetch_url(url)); // 非ブロッキング、即座に戻る
    }

    // 2. 結果を収集（送信順ではなく完了順）
    let mut results = Vec::new();
    while let Some(res) = set.join_next().await {
        results.push(res.unwrap());
    }
    results
}

async fn fetch_url(url: &str) -> String { /* 実装省略 */ "data".to_string() }

💡 メリット: Vec<JoinHandle> に比べてコード量を50%削減し、タスクキャンセル（set.abort_all()）をネイティブにサポートします。

💡 テクニック4：Async Drop の代替案－手動クリーンアップがより安全

なぜこのデザインが必要か？

Rust にはネイティブな Async Drop が存在しません。主な理由は Drop の「同期的な性質」にあります：スレッドがパニックした場合、ランタイムはリソースを同期的に解放する必要があります。一方、Async 操作はスケジューリングに依存するため、デッドロックを引き起こす可能性があります。そのため、コミュニティでは明示的な Async クリーンアップが推奨されています。これは本质的に「破棄ロジック」を Drop からユーザーが制御する Async 関数に移動するものです。

struct AsyncResource {
    conn: TcpStream, // Async クローズが必要なリソース
}

impl AsyncResource {
    // 解決策1：Async クリーンアップ関数を手動で呼び出し
    async fn close(&mut self) {
        self.conn.shutdown().await.unwrap(); // Async クローズロジック
    }
}

// 解決策2：クリーンアップを自動的にトリガーするガードパターン
struct ResourceGuard {
    inner: Option<AsyncResource>,
}

impl ResourceGuard {
    async fn drop_async(mut self) {
        if let Some(mut res) = self.inner.take() {
            res.close().await;
        }
    }
}

⚠️ 落とし穴回避: std::mem::forget を使用してクリーンアップをスキップしないでください。これによりリソースリークが発生します。

💡 テクニック5：Tokio ランタイムを最適化する－シナリオに合わせてスレッドモデルを設定

なぜこのデザインが必要か？

Tokio のデフォルトの「マルチスレッドワークスチーリング」モデルは、すべてのシナリオに適しているわけではありません。ランタイムのコアパラメータ（スレッド数、アロケータ、IO ドライバ）はパフォーマンスに直接影響を与えるため、IOバウンドまたはCPUバウンドのワークロードに合わせてカスタマイズする必要があります。

use tokio::runtime::{Builder, Runtime};

// シナリオ1：IOバウンド（例：APIサービス）－マルチスレッド＋io-uring
fn io_intensive_runtime() -> Runtime {
    Builder::new_multi_thread()
        .worker_threads(4) // スレッド数＝CPUコア数×2（IO待ち中に他のタスクをスケジューリング）
        .enable_io() // IOドライバを有効化（epoll/kqueue/io-uring）
        .enable_time() // タイマーを有効化（例：`sleep` 使用時）
        .build()
        .unwrap()
}

// シナリオ2：CPUバウンド（例：データ計算）－シングルスレッド＋IO無効
fn cpu_intensive_runtime() -> Runtime {
    Builder::new_current_thread()
        .enable_time()
        .build()
        .unwrap()
}

🔥 パフォーマンスノート: IOバウンドワークロードの場合は、epoll より30%以上高速な io-uring（Linux 5.1+）を使用しましょう。CPUバウンドワークロードの場合は、スレッド切り替えオーバーヘッドを回避するためにシングルスレッドを使用しましょう。

💡 テクニック6：Sync + Send の過剰使用を避ける－並行安全性制約を最小化

なぜこのデザインが必要か？

Sync（スレッド間での安全な共有）と Send（スレッド間での安全な転送）は Rust のコア並行トレイトですが、すべての Async タスクがこれらを必要とするわけではありません。例えば：

  • LocalSet 内のタスクは現在のスレッドでのみ実行されるため、Send は不要です。
  • シングルスレッドランタイム内の Future は Sync を必要としません。

これらのトレイトを過剰に使用すると、ジェネリック制約が不必要に厳しくなり、有効なユースケースが除外される可能性があります。

use tokio::task::LocalSet;

// `Send` を持たないタスク：現在のスレッドでのみ実行
async fn local_task() {
    let mut data = String::from("local");
    data.push_str(" data");
    println!("{}", data);
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let local_set = LocalSet::new();
    // 安全：`LocalSet` のタスクは `Send` を要求せず、非`Send`変数をキャプチャできる
    local_set.run_until(local_task()).await;
}

💡 テクニック: 非Sendタスクを許可するため、tokio::task::spawn の代わりに spawn_local を使用しましょう。ジェネリック制約では、T: Future + Send + Sync より T: Future を優先しましょう。

💡 テクニック7：Async サービスオーケストレーションに Tower を使用する－エレガントなミドルウェア実践

なぜこのデザインが必要か？

Tower は Async サービスのための「ミドルウェアフレームワーク」で、コアデザインは**Serviceトレイト＋コンビネータパターン**です。Async 開発の一般的な課題－汎用ロジック（タイムアウト、リトライ、レート制限）とビジネスコードの結合－を解決します。Layerトレイトを介して汎用ロジックをミドルウェアとしてカプセル化し、ビルディングブロックのように組み合わせることができ、「単一責任の原則」に沿った設計が可能です。

use tower::{Service, ServiceBuilder, service_fn, BoxError};
use tower::timeout::Timeout;
use tower::retry::Retry;
use std::time::Duration;

// 1. ビジネスロジック：リクエスト処理
async fn handle_request(req: String) -> Result<String, BoxError> {
    Ok(format!("response: {}", req))
}

// 2. ミドルウェアの組み合わせ：タイムアウト＋リトライ＋ビジネスロジック
fn build_service() -> impl Service<String, Response = String, Error = BoxError> {
    ServiceBuilder::new()
        .timeout(Duration::from_secs(3)) // タイムアウトミドルウェア
        .retry(tower::retry::Limited::new(2)) // 2回リトライ
        .service(service_fn(handle_request)) // ビジネスサービス
}

#[tokio::main]
async fn main() {
    let mut service = build_service();
    // 3. サービスの呼び出し
    let res = service.call("hello".to_string()).await.unwrap();
    println!("{}", res);
}

🔥 エコシステム: Tower は Axum や Hyper などのフレームワークと統合されており、Rust Async サービスの標準的なミドルウェアソリューションとなっています。

💡 テクニック8：Async Stream のバックプレッシャー処理－メモリ爆発を回避

なぜこのデザインが必要か？

Async Stream（例：futures::stream::Stream）は「async イテレータ」ですが、プロデューサー（データ生成側）の速度がコンシューマー（データ消費側）を上回ると、メモリ膨張が発生します。バックプレッシャーの核心は**「コンシューマーが Poll シグナルでプロデューサーの速度を制御する」** ことです。コンシューマーが繁忙な場合、Poll::Pending を返し、プロデューサーはデータ生成を一時停止します。

use futures::stream::{self, StreamExt};
use std::time::Duration;

// プロデューサー：1～1000 のストリームを生成
fn producer() -> impl futures::Stream<Item = u32> {
    stream::iter(1..1000)
}

// コンシューマー：バックプレッシャーを含む処理遅延をシミュレート
async fn consumer(mut stream: impl futures::Stream<Item = u32>) {
    while let Some(item) = stream.next().await {
        // 時間のかかる処理をシミュレート（実際はデータベース/ネットワーク IO）
        tokio::time::sleep(Duration::from_millis(10)).await;
        println!("processed: {}", item);
        // キーポイント：`next().await` が処理完了を待機し、間接的にプロデューサーの速度を制御
    }
}

#[tokio::main]
async fn main() {
    let stream = producer();
    consumer(stream).await;
}

⚠️ 落とし穴回避: stream::buffered を使用する場合は、無制限のキャッシュを防ぐため、適切なバッファサイズ（例：10）を設定しましょう。

💡 テクニック9：Unsafe Async の境界を制御する－Unsafe コードを最小化

なぜこのデザインが必要か？

Async 内で unsafe を使用するリスクは、Sync 内で使用する場合よりはるかに高いです：

  • Pin::new_unchecked を手動で呼び出すと、自己参照の安全性が破られる可能性があります。
  • async unsafe fn はスレッド間のデータ競合を引き起こす可能性があります。

Rust の設計理念は「unsafe コードは明示的にマークし、最小化すべき」です。そのため、Async 内の unsafe は厳格な境界制御が必要で、安全なラッパーを介してリスクを分離する必要があります。

use std::pin::Pin;
use std::future::Future;

// Unsafe な基盤実装：自己参照型 Future を手動で Pin
unsafe fn unsafe_pin_future<F: Future>(fut: F) -> Pin<Box<F>> {
    let boxed = Box::new(fut);
    // 安全性前提条件：呼び出し側は `fut` に自己参照がない、または移動しないことを保証
    Pin::new_unchecked(boxed)
}

// 安全なラッパー：外部使用から `unsafe` を隠し、前提条件が満たされることを保証
pub fn safe_pin_future<F: Future + Unpin>(fut: F) -> Pin<Box<F>> {
    // `Unpin` トレイトを使用して `fut` に自己参照がないことを保証し、unsafe の前提条件を満たす
    unsafe { unsafe_pin_future(fut) }
}

💡 原則: Async 内のすべての unsafe コードは、個別の関数に配置し、「安全性前提条件」を明確に文書化しましょう。

💡 テクニック10：Trace ツールチェーンを統合する－Async デバッグの「視点レンズ」

なぜこのデザインが必要か？

Async タスクのスケジューリングは「非連続的」です。1つのタスクが複数のスレッド間で切り替わる可能性があるため、従来のコールスタックはトレースに無効です。tracing＋opentelemetry ツールチェーンはイベント駆動型トレースに依存しています。スパン（span）を介してタスクのライフサイクルをマークし、スケジューリング、IO、エラーイベントを記録することで、「タスクの停滞」や「メモリリーク」などの問題を診断するのに役立ちます。

use tracing::{info, span, Level};
use tracing_subscriber::{prelude::*, EnvFilter};
use std::time::Duration;

async fn fetch_data() -> String {
    // 子スパン：サブタスクをマーク
    let span = span!(Level::INFO, "fetch_data");
    let _guard = span.enter();
    info!("sending request");
    tokio::time::sleep(Duration::from_secs(1)).await;
    info!("request completed");
    "ok".to_string()
}

#[tokio::main]
async fn main() {
    // Trace の初期化：コンソールに出力、環境変数でログをフィルタリング
    tracing_subscriber::registry()
        .with(EnvFilter::from_default_env())
        .with(tracing_subscriber::fmt::layer())
        .init();

    // スパンの作成：タスクの範囲をマーク
    let root_span = span!(Level::INFO, "main_task");
    let _guard = root_span.enter();

    info!("start fetching data");
    let data = fetch_data().await;
    info!("fetched data: {}", data);
}

🔥 ツール: タスクスケジューリングの可視化には tokio-console を、分散トレースの分析には Jaeger を使用しましょう。

まとめ

Rust Async 開発の核心は**「基盤となる制約を理解し、エコシステムツールを活用すること」** です。これら10のテクニックは、スケジューリング、メモリ、並行処理、デバッグにおける主要なシナリオをカバーしており、「どのように使うか」から「なぜそうなるか」まで理解を深めるのに役立ちます。実践では、Async が「万能薬」ではないことを忘れないでください。Sync/Async の境界を適切に分離することで、始めて高性能で安全なコードを記述できます。

Leapcell：最高のサーバーレス Web ホスティング

最後に、Rust サービスのデプロイに最適なプラットフォームである Leapcell を推奨します。

🚀 お気に入りの言語で開発

JavaScript、Python、Go、Rust などを使って、手軽に開発できます。

🌍 無料で無制限のプロジェクトをデプロイ

使用した分だけ料金を支払うだけで、受信リクエストには料金がかかりません。

⚡ 従量課金制、隠れたコストなし

アイドル料金はなく、シームレスにスケーラビリティを向上できます。

📖 ドキュメントを探索

🔹 Twitter でフォロー：@LeapcellHQ

Discussion