Mempercepat download objek Cloud Storage

Mendownload objek besar di Cloud Storage secara berurutan melalui satu koneksi dapat menyebabkan bandwidth jaringan yang tersedia tidak digunakan. Dengan membagi objek menjadi segmen rentang byte yang disebut stripe dan membaca segmen tersebut secara serentak, Anda dapat meningkatkan throughput download untuk objek besar menggunakan Library Klien Cloud untuk Rust.

Dokumen ini menunjukkan cara menyisipkan objek pengujian besar menggunakan komposisi objek, mendownload rentang byte secara serentak dengan ReadRange::segment, dan menulis stripe ke file lokal pada offset tetap. Jika Anda baru menggunakan library klien, lihat Menggunakan Cloud Storage dengan Rust terlebih dahulu.

Sebelum memulai

Sebelum menjalankan kode dalam dokumen ini, selesaikan prasyarat berikut:

• Buat atau pilih projectGoogle Cloud dengan penagihan diaktifkan.
• Buat bucket Cloud Storage pengujian khusus di project Anda.
• Konfigurasikan Kredensial Default Aplikasi untuk lingkungan Anda.
• Beri akun utama Anda peran Identity and Access Management (IAM) dengan izin untuk membaca, menulis, menyusun, dan menghapus objek di bucket, seperti Storage Admin (roles/storage.admin).

Catatan: Menjalankan langkah penyemaian penuh akan membuat beberapa objek pengujian hingga masing-masing 32 GiB (total lebih dari 63 GiB). Hapus objek pengujian dan file output lokal segera setelah Anda selesai melakukan pengujian untuk menghindari biaya penyimpanan yang berkelanjutan.

Dependensi

Tambahkan library klien Cloud Storage dan crate pendukung ke file Cargo.toml Anda:

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

Membuat objek pengujian besar

Untuk menguji download bergaris tanpa mengupload gigabyte data melalui jaringan, upload objek seed kecil dan gunakan operasi compose Cloud Storage untuk menggabungkan salinan di sisi server. Di dalam fungsi seed asinkron yang menerima client, control, dan bucket_name dalam format projects/_/buckets/BUCKET_ID, selesaikan langkah-langkah berikut:

1. Untuk mereferensikan objek yang ada dalam permintaan compose, impor SourceObject:

   use google_cloud_storage::model::compose_object_request::SourceObject;

2. Upload objek seed 1 MiB bernama 1MiB.txt:

   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
   );

3. Untuk membuat objek 32 MiB bernama 32MiB.txt, gabungkan 32 salinan 1MiB.txt di server:

   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)
   );

4. Untuk membuat objek 1 GiB bernama 1GiB.txt, gabungkan 32 salinan 32MiB.txt:

   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)
   );

5. Untuk membuat objek pengujian pangkat dua dari 1GiB.txt (2GiB.txt, 4GiB.txt, 8GiB.txt, 16GiB.txt, dan 32GiB.txt), ulangi operasi compose:

   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)
       );
   }

Mengoordinasikan download stripe serentak

Untuk mendownload objek secara paralel, buat kueri ukuran dan generasi-nya, alokasikan file tujuan lokal terlebih dahulu, dan buat satu future per stripe. Di dalam fungsi download yang menerima client, control, bucket_name, object_name, stripe_size, dan destination, selesaikan langkah-langkah berikut:

1. Ambil metadata objek menggunakan StorageControl::get_object agar Anda dapat memeriksa ukuran total objek dan mengunci pembacaan berikutnya ke satu generasi:

   let metadata = control
       .get_object()
       .set_bucket(bucket_name)
       .set_object(object_name)
       .send()
       .await?;

2. Buat file tujuan dan alokasikan terlebih dahulu ke ukuran objek yang tepat sehingga tugas worker dapat menulis pada offset byte independen:

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

3. Hitung setiap offset stripe menggunakan stripe_size sebagai limit dan buat write_stripe future per segmen, dengan menambahkan stripe parsial akhir jika ukuran objek tidak dapat dibagi rata oleh 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,
       ))
   }

4. Jalankan semua stripe future secara serentak dengan futures::future::join_all dan propagasi error apa pun:

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

Mendownload dan menulis stripe satu per satu

Setiap tugas write_stripe membaca satu rentang byte dari Cloud Storage dan menulis potongan yang diterima pada offset yang sesuai di pengendali file lokal bersama. Untuk menerapkan write_stripe, selesaikan langkah-langkah berikut:

1. Impor ReadRange dan tentukan fungsi 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;

2. Buat builder permintaan ReadObject dengan Storage::read_object, dan kirim permintaan setelah mengonfigurasi opsi pembuatan dan rentang baca di langkah-langkah berikut:

   let mut reader = client
       .read_object(&metadata.bucket, &metadata.name)
       .send()
       .await?;

3. Untuk menyematkan semua stripe ke pembuatan objek yang sama dan mencegah pembacaan yang tidak konsisten jika proses lain menimpa objek, panggil set_generation pada builder permintaan sebelum memanggil send:

   .set_generation(metadata.generation)

4. Untuk membatasi pembacaan ke rentang byte stripe, panggil set_read_range dengan ReadRange::segment(offset, limit) pada builder permintaan:

   .set_read_range(ReadRange::segment(offset, limit))

5. Streaming potongan dari pembaca dan tulis setiap potongan ke file tujuan di current_pos di dalam tokio::task::spawn_blocking. Karena library standar Rust tidak memiliki metode penulisan posisi lintas platform, contoh ini menggunakan FileExt::write_all_at di Unix dan FileExt::seek_write di Windows:

   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;
   }

Pertimbangan performa

Beberapa faktor menentukan throughput efektif download bergaris:

• Kecepatan disk lokal. Menulis potongan ke penyimpanan lokal yang lambat dapat menjadi bottleneck. Menggunakan penyimpanan SSD lokal dengan throughput tinggi membantu menjaga I/O disk tetap lebih cepat daripada pembacaan jaringan.
• Kapasitas CPU VM. Cloud Storage mengenkripsi data objek dalam pengiriman, sehingga klien mendekripsi setiap potongan yang diterimanya. Pastikan VM Anda memiliki vCPU yang cukup untuk menangani dekripsi TLS serentak di seluruh stripe.
• Lokasi bucket dan VM. Mendownload di seluruh region akan menimbulkan latensi jaringan area luas dan membatasi throughput. Tempatkan VM Anda di region yang sama dengan bucket Anda jika memungkinkan.

Anda juga dapat bereksperimen dengan menyesuaikan stripe_size untuk menghindari sisa stripe yang sangat kecil atau membuat terlalu banyak permintaan rentang kecil.

Kode lengkap

Contoh berikut menggabungkan fungsi seed, download, dan write_stripe menjadi modul lengkap:

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(())
}

Saat Anda menjalankan download pada objek 32 MiB dengan stripe_size 1 MiB, outputnya mirip dengan berikut ini, yang melaporkan ukuran yang didownload, waktu yang berlalu, jumlah stripe, dan bandwidth efektif:

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

Pembersihan

Agar akun Google Cloud Anda tidak dikenai biaya untuk objek pengujian yang dibuat oleh seed, hapus objek pengujian dan hapus bucket pengujian Anda setelah selesai. Hapus juga file tujuan lokal.