AWS SDK for Rust での同時実行

AWS SDK for Rust は同時実行制御を提供しませんが、ユーザーは独自の同時実行制御を実装するための多くのオプションがあります。

用語

このテーマに関連する用語は混同されやすく、もともと別々の概念を表していたにもかかわらず、同義語となった用語もあります。このガイドでは、以下の用語を定義します。

シンプルな例

最初の例は、シーケンシャルプログラムです。後の例では、このコードを同時実行手法で変更します。後の例では、同じ build_client_and_list_objects_to_download() メソッドを再利用して、main() 内で変更を加えます。次のコマンドを実行して、プロジェクトに依存関係を追加します。

次のタスク例では、Amazon Simple Storage Service バケット内のすべてのファイルをダウンロードしています。

use aws_sdk_s3::{Client, Error};
const EXAMPLE_BUCKET: &str = "amzn-s3-demo-bucket";  // Update to name of bucket you own.

// This initialization function won't be reproduced in
// examples following this one, in order to save space.
async fn build_client_and_list_objects_to_download() -> (Client, Vec<String>) {
    let cfg = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await;
    let client = Client::new(&cfg);
    let objects_to_download: Vec<_> = client
        .list_objects_v2()
        .bucket(EXAMPLE_BUCKET)
        .send()
        .await
        .expect("listing objects succeeds")
        .contents()
        .into_iter()
        .flat_map(aws_sdk_s3::types::Object::key)
        .map(ToString::to_string)
        .collect();
         
    (client, objects_to_download)
}

#[tokio::main]
async fn main() {
    let (client, objects_to_download) =
        build_client_and_list_objects_to_download().await;
    
    for object in objects_to_download {
        let res = client
            .get_object()
            .key(&object)
            .bucket(EXAMPLE_BUCKET)
            .send()
            .await
            .expect("get_object succeeds");
        let body = res.body.collect().await.expect("reading body succeeds").into_bytes();
        std::fs::write(object, body).expect("write succeeds");
    }
}

最新のコンピュータのアーキテクチャを使用すれば、このプログラムをはるかに効率的なものに書き直すことができます。これについては後の例で説明しますが、まずは概念についてさらにいくつか説明します。

所有権と可変性

Rust のそれぞれの値には単一の所有者を持ちます。所有者が範囲から外れると、その所有者が所有するすべての値も削除されます。所有者は、1 つ以上の値に対する変更不可能な参照または 1 つの変更可能な参照を提供できます。Rust コンパイラは、参照がその所有者を越えて存在しないことを保証します。

複数のタスクが同じリソースに可変的にアクセスする必要がある場合は、追加の計画と設計が必要になります。シーケンシャルコンピューティングでは、各タスクは順番に実行されるため、競合することなく可変的に同じリソースにアクセスできます。ただし、同時コンピューティングでは、タスクは任意の順序で同時に実行できます。したがって、複数の変更可能な参照が不可能であることをコンパイラに証明するため (または、少なくともクラッシュするように)、さらに多くの対策が必要です。

Rust 標準ライブラリには、これを実現するための多くのツールが用意されています。これらのトピックの詳細については、「Rust Programming Language」書籍の「変数と可変性」および「所有権を理解する」を参照してください。

その他の用語

以下は「同期オブジェクト」の一覧です。これらは、コンパイラに対して同時実行プログラムが所有権ルールに反しないことを保証するために必要な手段です。

標準ライブラリ同期オブジェクト:

Tokio 同期オブジェクト:

AWS SDK は async ランタイムに依存しないことを想定していますが、特定のケースでは tokio 同期オブジェクトを使用することをお勧めします。

例をより効率的に書き直す (シングルスレッド同時実行)

次の変更例では、futures_util::future::join_all を使用して すべての get_object リクエストを同時に実行します。次のコマンドを実行して、プロジェクトに新しい依存関係を追加します。

#[tokio::main]
async fn main() {
    let (client, objects_to_download) =
        build_client_and_list_objects_to_download().await;
        
    let get_object_futures = objects_to_download.into_iter().map(|object| {
        let req = client
            .get_object()
            .key(&object)
            .bucket(EXAMPLE_BUCKET);

        async {
            let res = req
                .send()
                .await
                .expect("get_object succeeds");
            let body = res.body.collect().await.expect("body succeeds").into_bytes();
           // Note that we MUST use the async runtime's preferred way
           // of writing files. Otherwise, this call would block,
           // potentially causing a deadlock.
            tokio::fs::write(object, body).await.expect("write succeeds");
        }
    });

    futures_util::future::join_all(get_object_futures).await;
}

これは同時実行のメリットを享受する最も簡単な方法ですが、最初はわかりにくい問題が一部あります。

これらの問題を両方とも解決するには、一度に送信するリクエスト量を制限する必要があります。これを行うには、tokio セマフォを使用します。

use std::sync::Arc;
use tokio::sync::Semaphore;
const CONCURRENCY_LIMIT: usize = 50; 

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (client, objects_to_download) =
        build_client_and_list_objects_to_download().await;
    let concurrency_semaphore = Arc::new(Semaphore::new(CONCURRENCY_LIMIT));

    let get_object_futures = objects_to_download.into_iter().map(|object| {
        // Since each future needs to acquire a permit, we need to clone
        // the Arc'd semaphore before passing it in.
        let semaphore = concurrency_semaphore.clone();
        // We also need to clone the client so each task has its own handle.
        let client = client.clone();
        async move {
            let permit = semaphore
                .acquire()
                .await
                .expect("we'll get a permit if we wait long enough");
            let res = client
                .get_object()
                .key(&object)
                .bucket(EXAMPLE_BUCKET)
                .send()
                .await
                .expect("get_object succeeds");
            let body = res.body.collect().await.expect("body succeeds").into_bytes();
            tokio::fs::write(object, body).await.expect("write succeeds");
            std::mem::drop(permit);
        }
    });

    futures_util::future::join_all(get_object_futures).await;
}

リクエストの作成を async ブロックに移動することにより、潜在的なメモリ使用量の問題を修正しました。この方法により、リクエストは送信するタイミングまで作成されません。

また、転送中のリクエストを CONCURRENCY_LIMIT に制限することにより、一度に大量のリクエストが送信される問題を修正しました。

例をより効率的に書き直す (マルチスレッド同時実行)

前の 2 つの例では、リクエストを同時に実行しました。これは同期的に実行するよりも効率的ですが、マルチスレッドを使用することにより、さらに効率が高まります。tokio でこれを行うには、それらを別々のタスクとして生成する必要があります。

次のコマンドを実行して、プロジェクトに新しい依存関係を追加します。

// Set this based on the amount of cores your target machine has.
const THREADS: usize = 8; 

#[tokio::main(flavor = "multi_thread")]
async fn main() {
    let (client, objects_to_download) =
        build_client_and_list_objects_to_download().await;
    let concurrency_semaphore = Arc::new(Semaphore::new(THREADS));

    let get_object_task_handles = objects_to_download.into_iter().map(|object| {
        // Since each future needs to acquire a permit, we need to clone
        // the Arc'd semaphore before passing it in.
        let semaphore = concurrency_semaphore.clone();
        // We also need to clone the client so each task has its own handle.
        let client = client.clone();
        
        // Note this difference! We're using `tokio::task::spawn` to
        // immediately begin running these requests.
        tokio::task::spawn(async move {
            let permit = semaphore
                .acquire()
                .await
                .expect("we'll get a permit if we wait long enough");
            let res = client
                .get_object()
                .key(&object)
                .bucket(EXAMPLE_BUCKET)
                .send()
                .await
                .expect("get_object succeeds");
            let body = res.body.collect().await.expect("body succeeds").into_bytes();
            tokio::fs::write(object, body).await.expect("write succeeds");
            std::mem::drop(permit);
        })
    });

    futures_util::future::join_all(get_object_task_handles).await;
}

作業をタスクに分割するのは複雑な場合があります。通常、I/O (入力/出力) の実行はブロックされます。ランタイムは、長時間実行されるタスクのニーズと短時間実行されるタスクのニーズのバランスをとるのが難しい場合があります。どのランタイムを選択する場合でも、作業をタスクに分割する最も効率的な方法についての推奨事項を必ず確認してください。tokio ランタイムの推奨事項については、「モジュール tokio::task」を参照してください。

マルチスレッドアプリケーションのデバッグ

同時に実行されるタスクは、任意の順序で実行できます。そのため、同時プログラムのログは非常に読みにくい場合があります。SDK for Rust では、tracing ロギングシステムを使用することをお勧めします。実行中であっても、ログを特定のタスクでグループ化できます。ガイダンスについては、「AWS SDK for Rust でのログ記録の設定と使用」を参照してください。

ロックされたタスクを特定するための非常に便利なツールとして tokio-console があります。これは、非同期 Rust プログラムの診断およびデバッグツールです。プログラムを実装して実行し、次に tokio-console アプリケーションを実行すると、プログラムが実行されているタスクのライブビューが表示されます。このビューには、タスクが共有リソースの取得を待機するのにかかった時間や、ポーリングされた時間などの有用な情報が含まれています。