Google は AI 技術を使用して、コンテンツをご希望の言語に翻訳しています。AI 翻訳には誤りが含まれる場合があります。

Cloud Storage オブジェクトのダウンロードを高速化する

Cloud Storage で大きなオブジェクトを単一の接続で順番にダウンロードすると、使用可能なネットワーク帯域幅が未使用のままになることがあります。オブジェクトをストライプと呼ばれるバイト範囲セグメントに分割し、それらのセグメントを同時に読み取ることで、Rust 用 Cloud クライアントライブラリを使用して、大きなオブジェクトのダウンロードスループットを向上させることができます。

このドキュメントでは、オブジェクトの作成を使用して大規模なテストオブジェクトをシードし、ReadRange::segment と同時にバイト範囲をダウンロードして、固定オフセットでローカルファイルにストライプを書き込む方法について説明します。クライアントライブラリを初めて使用する場合は、まず Rust で Cloud Storage を使用するをご覧ください。

始める前に

このドキュメントのコードを実行する前に、次の前提条件を満たしてください。

課金が有効になっている Google Cloud プロジェクトを作成または選択します。
プロジェクトに専用のテスト用 Cloud Storage バケットを作成します。
使用する環境のアプリケーションのデフォルト認証情報を構成します。
バケット内のオブジェクトの読み取り、書き込み、作成、削除を行う権限（ストレージ管理者（roles/storage.admin）など）を持つ Identity and Access Management（IAM）ロールをプリンシパルに付与します。

依存関係

Cloud Storage クライアントライブラリとサポートクレートを Cargo.toml ファイルに追加します。

cargo add google-cloud-storage google-cloud-gax anyhow bytes futures
cargo add tokio --features macros,rt-multi-thread

大きなテストオブジェクトを作成する

ネットワーク経由でギガバイト単位のデータをアップロードせずにストライプダウンロードをテストするには、小さなシードオブジェクトをアップロードし、Cloud Storage のcompose オペレーションを使用してサーバーサイドでコピーを連結します。client、control、bucket_name を projects/_/buckets/BUCKET_ID 形式で受け入れる非同期 seed 関数内で、次の手順を完了します。

作成リクエストで既存のオブジェクトを参照するには、SourceObject をインポートします。
```
use google_cloud_storage::model::compose_object_request::SourceObject;
```

1MiB.txt という名前の 1 MiB のシードオブジェクトをアップロードします。

let buffer = String::from_iter(('a'..='z').cycle().take(1024 * 1024));
let seed = client
    .write_object(bucket_name, "1MiB.txt", bytes::Bytes::from_owner(buffer))
    .set_if_generation_match(0)
    .send_unbuffered()
    .await?;
println!(
    "Uploaded object {}, size={}KiB",
    seed.name,
    seed.size / 1024
);

32MiB.txt という名前の 32 MiB のオブジェクトを作成するには、サーバー上で 1MiB.txt の 32 個のコピーを連結します。

let seed_32 = control
    .compose_object()
    .set_destination(Object::new().set_bucket(bucket_name).set_name("32MiB.txt"))
    .set_source_objects((0..32).map(|_| {
        SourceObject::new()
            .set_name(&seed.name)
            .set_generation(seed.generation)
    }))
    .set_if_generation_match(0)
    .with_idempotency(true)
    .send()
    .await?;
println!(
    "Created object {}, size={}MiB",
    seed.name,
    seed.size / (1024 * 1024)
);

1GiB.txt という名前の 1 GiB オブジェクトを作成するには、32MiB.txt の 32 個のコピーを連結します。

let seed_1024 = control
    .compose_object()
    .set_destination(Object::new().set_bucket(bucket_name).set_name("1GiB.txt"))
    .set_source_objects((0..32).map(|_| {
        SourceObject::new()
            .set_name(&seed_32.name)
            .set_generation(seed_32.generation)
    }))
    .set_if_generation_match(0)
    .with_idempotency(true)
    .send()
    .await?;
println!(
    "Created object {}, size={}MiB",
    seed.name,
    seed.size / (1024 * 1024)
);

1GiB.txt（2GiB.txt、4GiB.txt、8GiB.txt、16GiB.txt、32GiB.txt）から 2 のべき乗のテストオブジェクトを作成するには、作成オペレーションを繰り返します。

for s in [2, 4, 8, 16, 32] {
    let name = format!("{s}GiB.txt");
    let target = control
        .compose_object()
        .set_destination(Object::new().set_bucket(bucket_name).set_name(&name))
        .set_source_objects((0..s).map(|_| {
            SourceObject::new()
                .set_name(&seed_1024.name)
                .set_generation(seed_1024.generation)
        }))
        .set_if_generation_match(0)
        .with_idempotency(true)
        .send()
        .await?;
    println!(
        "Created object {} size={} MiB",
        target.name,
        target.size / (1024 * 1024)
    );
}

同時ストライプダウンロードを調整する

オブジェクトを並列でダウンロードするには、サイズと世代をクエリし、ローカルの宛先ファイルを事前に割り当てて、ストライプごとに 1 つの future を生成します。client、control、bucket_name、object_name、stripe_size、destination を受け入れる download 関数内で、次の手順を完了します。

StorageControl::get_object を使用してオブジェクトメタデータを取得し、オブジェクトの合計サイズを検査して、後続の読み取りを単一の世代にロックできるようにします。
```
let metadata = control
    .get_object()
    .set_bucket(bucket_name)
    .set_object(object_name)
    .send()
    .await?;
```
宛先ファイルを作成し、ワーカータスクが独立したバイトオフセットで書き込めるように、正確なオブジェクトサイズに事前割り当てします。
```
let file = Arc::new(std::fs::File::create(destination)?);
file.set_len(metadata.size as u64)?;
```

stripe_size を limit として使用して各ストライプオフセットを計算し、セグメントごとに write_stripe フューチャーを作成します。オブジェクトサイズが limit で割り切れない場合は、最後の部分ストライプを追加します。

let size = metadata.size as u64;
let limit = stripe_size as u64;
let count = size / limit;
let mut stripes = (0..count)
    .map(|i| write_stripe(client.clone(), &file, i * limit, limit, &metadata))
    .collect::<Vec<_>>();
if !size.is_multiple_of(limit) {
    stripes.push(write_stripe(
        client.clone(),
        &file,
        count * limit,
        limit,
        &metadata,
    ))
}

futures::future::join_all を使用してすべてのストライプフューチャーを同時に実行し、エラーを伝播します。
```
futures::future::join_all(stripes)
    .await
    .into_iter()
    .collect::<anyhow::Result<Vec<_>>>()?;
```

個々のストライプをダウンロードして書き込む

各 write_stripe タスクは、Cloud Storage から 1 つのバイト範囲を読み取り、受信したチャンクを共有ローカルファイル操作の対応するオフセットに書き込みます。write_stripe を実装する手順は次のとおりです。

ReadRange をインポートし、write_stripe 関数を定義します。

async fn write_stripe(
    client: Storage,
    file: &Arc<std::fs::File>,
    offset: u64,
    limit: u64,
    metadata: &Object,
) -> anyhow::Result<()> {
    use google_cloud_storage::model_ext::ReadRange;

Storage::read_object を使用して ReadObject リクエストビルダーを作成し、次の手順で生成オプションと読み取り範囲オプションを構成してからリクエストを送信します。
```
let mut reader = client
    .read_object(&metadata.bucket, &metadata.name)
    .send()
    .await?;
```
すべてのストライプを同じオブジェクト世代に固定し、別のプロセスがオブジェクトを上書きした場合に不整合な読み取りを防ぐには、send を呼び出す前にリクエストビルダーで set_generation を呼び出します。
```
.set_generation(metadata.generation)
```
読み取りをストライプのバイト範囲に制限するには、リクエストビルダーで ReadRange::segment(offset, limit) を指定して set_read_range を呼び出します。
```
.set_read_range(ReadRange::segment(offset, limit))
```

リーダーからチャンクをストリーミングし、各チャンクを tokio::task::spawn_blocking 内の current_pos にある宛先ファイルに書き込みます。Rust の標準ライブラリにはクロスプラットフォームの位置指定書き込みメソッドがないため、この例では Unix で FileExt::write_all_at、Windows で FileExt::seek_write を使用しています。

let mut current_pos = offset;
while let Some(b) = reader.next().await.transpose()? {
    let chunk_len = b.len() as u64;
    let handle = file.clone();
    #[cfg(unix)]
    tokio::task::spawn_blocking(move || {
        use std::os::unix::fs::FileExt;
        handle.write_all_at(&b, current_pos)
    })
    .await??;
    #[cfg(windows)]
    tokio::task::spawn_blocking(move || {
        use std::os::windows::fs::FileExt;
        handle.seek_write(&b, current_pos)
    })
    .await??;
    current_pos += chunk_len;
}

パフォーマンスに関する注意事項

ストライプダウンロードの有効なスループットは、次のいくつかの要因によって決まります。

ローカルディスクの速度。低速なローカルストレージにチャンクを書き込むと、ボトルネックになる可能性があります。高スループットのローカル SSD ストレージを使用すると、ディスク I/O をネットワーク読み取りよりも優先できます。
VM CPU 容量。Cloud Storage は転送中のオブジェクトデータを暗号化するため、クライアントは受信したすべてのチャンクを復号します。VM に、ストライプ間で同時 TLS 復号を処理するのに十分な vCPU があることを確認します。
バケットと VM のロケーション。リージョン間でダウンロードすると、広域ネットワークのレイテンシが発生し、スループットが制限されます。可能であれば、バケットと同じリージョンに VM を配置します。

stripe_size のチューニングを試して、非常に小さな残りのストライプが残ったり、小さな範囲リクエストが多数作成されたりしないようにすることもできます。

コードを補完する

次の例では、seed、download、write_stripe の各関数を組み合わせて完全なモジュールにしています。

use std::sync::Arc;

use google_cloud_gax::options::RequestOptionsBuilder;
use google_cloud_storage::client::Storage;
use google_cloud_storage::client::StorageControl;
use google_cloud_storage::model::Object;

async fn seed(client: Storage, control: StorageControl, bucket_name: &str) -> anyhow::Result<()> {
    use google_cloud_storage::model::compose_object_request::SourceObject;

    let buffer = String::from_iter(('a'..='z').cycle().take(1024 * 1024));
    let seed = client
        .write_object(bucket_name, "1MiB.txt", bytes::Bytes::from_owner(buffer))
        .set_if_generation_match(0)
        .send_unbuffered()
        .await?;
    println!(
        "Uploaded object {}, size={}KiB",
        seed.name,
        seed.size / 1024
    );

    let seed_32 = control
        .compose_object()
        .set_destination(Object::new().set_bucket(bucket_name).set_name("32MiB.txt"))
        .set_source_objects((0..32).map(|_| {
            SourceObject::new()
                .set_name(&seed.name)
                .set_generation(seed.generation)
        }))
        .set_if_generation_match(0)
        .with_idempotency(true)
        .send()
        .await?;
    println!(
        "Created object {}, size={}MiB",
        seed.name,
        seed.size / (1024 * 1024)
    );

    let seed_1024 = control
        .compose_object()
        .set_destination(Object::new().set_bucket(bucket_name).set_name("1GiB.txt"))
        .set_source_objects((0..32).map(|_| {
            SourceObject::new()
                .set_name(&seed_32.name)
                .set_generation(seed_32.generation)
        }))
        .set_if_generation_match(0)
        .with_idempotency(true)
        .send()
        .await?;
    println!(
        "Created object {}, size={}MiB",
        seed.name,
        seed.size / (1024 * 1024)
    );

    for s in [2, 4, 8, 16, 32] {
        let name = format!("{s}GiB.txt");
        let target = control
            .compose_object()
            .set_destination(Object::new().set_bucket(bucket_name).set_name(&name))
            .set_source_objects((0..s).map(|_| {
                SourceObject::new()
                    .set_name(&seed_1024.name)
                    .set_generation(seed_1024.generation)
            }))
            .set_if_generation_match(0)
            .with_idempotency(true)
            .send()
            .await?;
        println!(
            "Created object {} size={} MiB",
            target.name,
            target.size / (1024 * 1024)
        );
    }

    Ok(())
}

async fn download(
    client: Storage,
    control: StorageControl,
    bucket_name: &str,
    object_name: &str,
    stripe_size: usize,
    destination: &str,
) -> anyhow::Result<()> {
    let metadata = control
        .get_object()
        .set_bucket(bucket_name)
        .set_object(object_name)
        .send()
        .await?;

    let file = Arc::new(std::fs::File::create(destination)?);
    file.set_len(metadata.size as u64)?;
    let start = std::time::Instant::now();

    let size = metadata.size as u64;
    let limit = stripe_size as u64;
    let count = size / limit;
    let mut stripes = (0..count)
        .map(|i| write_stripe(client.clone(), &file, i * limit, limit, &metadata))
        .collect::<Vec<_>>();
    if !size.is_multiple_of(limit) {
        stripes.push(write_stripe(
            client.clone(),
            &file,
            count * limit,
            limit,
            &metadata,
        ))
    }

    futures::future::join_all(stripes)
        .await
        .into_iter()
        .collect::<anyhow::Result<Vec<_>>>()?;

    let elapsed = start.elapsed();
    let mib = metadata.size as f64 / (1024.0 * 1024.0);
    let bw = mib / elapsed.as_secs_f64();
    println!(
        "Completed {mib:.2} MiB download in {elapsed:?}, using {count} stripes, effective bandwidth = {bw:.2} MiB/s"
    );

    Ok(())
}

async fn write_stripe(
    client: Storage,
    file: &Arc<std::fs::File>,
    offset: u64,
    limit: u64,
    metadata: &Object,
) -> anyhow::Result<()> {
    use google_cloud_storage::model_ext::ReadRange;
    let mut reader = client
        .read_object(&metadata.bucket, &metadata.name)
        .set_generation(metadata.generation)
        .set_read_range(ReadRange::segment(offset, limit))
        .send()
        .await?;
    let mut current_pos = offset;
    while let Some(b) = reader.next().await.transpose()? {
        let chunk_len = b.len() as u64;
        let handle = file.clone();
        #[cfg(unix)]
        tokio::task::spawn_blocking(move || {
            use std::os::unix::fs::FileExt;
            handle.write_all_at(&b, current_pos)
        })
        .await??;
        #[cfg(windows)]
        tokio::task::spawn_blocking(move || {
            use std::os::windows::fs::FileExt;
            handle.seek_write(&b, current_pos)
        })
        .await??;
        current_pos += chunk_len;
    }
    Ok(())
}

1 MiB の stripe_size を使用して 32 MiB のオブジェクトで download を実行すると、出力は次のようになります。ダウンロードされたサイズ、経過時間、ストライプ数、有効帯域幅が報告されます。

Completed 32.00 MiB download in 412.5ms, using 32 stripes, effective bandwidth = 77.58 MiB/s

クリーンアップ

seed によって作成されたテストオブジェクトに対して Google Cloud アカウントに課金されないようにするには、テストが完了したらテストオブジェクトを削除し、テストバケットを削除します。ローカルの宛先ファイルも削除します。