Downloads von Cloud Storage-Objekten beschleunigen

Wenn Sie ein großes Objekt in Cloud Storage sequenziell über eine einzelne Verbindung herunterladen, wird die verfügbare Netzwerkbandbreite möglicherweise nicht vollständig genutzt. Wenn Sie ein Objekt in Bytebereichssegmente (sogenannte Stripes) aufteilen und diese Segmente gleichzeitig lesen, können Sie den Downloaddurchsatz für große Objekte mit den Cloud-Clientbibliotheken für Rust verbessern.

In diesem Dokument wird beschrieben, wie Sie große Testobjekte mithilfe der Objektzusammensetzung erstellen, Bytebereiche gleichzeitig mit ReadRange::segment herunterladen und Stripes mit festen Offsets in eine lokale Datei schreiben. Wenn Sie die Clientbibliothek noch nicht kennen, lesen Sie zuerst Cloud Storage mit Rust verwenden.

Hinweis

Bevor Sie den Code in diesem Dokument ausführen, müssen die folgenden Voraussetzungen erfüllt sein:

• Erstellen oder wählen Sie ein Google Cloud -Projekt mit aktivierter Abrechnung aus.
• Erstellen Sie einen dedizierten Cloud Storage-Bucket für Tests in Ihrem Projekt.
• Konfigurieren Sie Standardanmeldedaten für Anwendungen für Ihre Umgebung.
• Weisen Sie Ihrem Hauptkonto eine IAM-Rolle (Identity and Access Management) mit der Berechtigung zum Lesen, Schreiben, Erstellen und Löschen von Objekten im Bucket zu, z. B. Storage-Administrator (roles/storage.admin).

Hinweis: Wenn Sie den vollständigen Seeding-Schritt ausführen, werden mehrere Testobjekte mit jeweils bis zu 32 GiB (insgesamt über 63 GiB) erstellt. Löschen Sie die Testobjekte und die lokale Ausgabedatei, sobald Sie die Tests abgeschlossen haben, um laufende Speicherkosten zu vermeiden.

Abhängigkeiten

Fügen Sie die Cloud Storage-Clientbibliothek und die unterstützenden Crates der Datei Cargo.toml hinzu:

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

Große Testobjekte erstellen

Wenn Sie gestreifte Downloads testen möchten, ohne Gigabyte an Daten über das Netzwerk hochzuladen, laden Sie ein kleines Seed-Objekt hoch und verwenden Sie den Compose-Vorgang von Cloud Storage, um Kopien serverseitig zu verketten. Führen Sie in einer asynchronen seed-Funktion, die client, control und bucket_name im Format projects/_/buckets/BUCKET_ID akzeptiert, die folgenden Schritte aus:

1. Importieren Sie SourceObject, um in Compose-Anfragen auf vorhandene Objekte zu verweisen:

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

2. Laden Sie ein 1 MiB großes Seed-Objekt mit dem Namen 1MiB.txt hoch:

   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. Wenn Sie ein 32-MiB-Objekt mit dem Namen 32MiB.txt erstellen möchten, verketten Sie 32 Kopien von 1MiB.txt auf dem 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. Wenn Sie ein 1-GiB-Objekt mit dem Namen 1GiB.txt erstellen möchten, verketten Sie 32 Kopien von 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. Wenn Sie Testobjekte mit Zweierpotenzen aus 1GiB.txt (2GiB.txt, 4GiB.txt, 8GiB.txt, 16GiB.txt und 32GiB.txt) erstellen möchten, wiederholen Sie den Erstellungsvorgang:

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

Gleichzeitige Stripe-Downloads koordinieren

Wenn Sie ein Objekt parallel herunterladen möchten, fragen Sie die Größe und Generierung ab, weisen Sie die lokale Zieldatei vorab zu und erstellen Sie ein Future pro Stripe. Führen Sie in einer download-Funktion, die client, control, bucket_name, object_name, stripe_size und destination akzeptiert, die folgenden Schritte aus:

1. Rufen Sie die Objektmetadaten mit StorageControl::get_object ab, um die Gesamtgröße des Objekts zu prüfen und nachfolgende Lesevorgänge auf eine einzelne Generierung zu beschränken:

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

2. Erstellen Sie die Zieldatei und weisen Sie ihr die genaue Objektgröße zu, damit Worker-Aufgaben an unabhängigen Byte-Offsets schreiben können:

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

3. Berechnen Sie jeden Stripe-Offset mit stripe_size als limit und erstellen Sie ein write_stripe-Future pro Segment. Hängen Sie einen letzten Teil-Stripe an, wenn die Objektgröße nicht gleichmäßig durch limit teilbar ist:

   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. Führen Sie alle Stripe-Futures gleichzeitig mit futures::future::join_all aus und geben Sie alle Fehler weiter:

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

Einzelnen Stripe herunterladen und schreiben

Jede write_stripe-Aufgabe liest einen einzelnen Bytebereich aus Cloud Storage und schreibt die empfangenen Blöcke mit dem entsprechenden Offset in das freigegebene lokale Dateihandle. So implementierst du write_stripe:

1. Importieren Sie ReadRange und definieren Sie die Funktion 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. Erstellen Sie mit Storage::read_object einen ReadObject-Anfrage-Builder und senden Sie die Anfrage, nachdem Sie die Optionen für die Generierung und den Lesebereich in den folgenden Schritten konfiguriert haben:

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

3. Wenn Sie alle Stripes an dieselbe Objektgeneration anpinnen und inkonsistente Lesevorgänge verhindern möchten, falls ein anderer Prozess das Objekt überschreibt, rufen Sie set_generation im Request Builder auf, bevor Sie send aufrufen:

   .set_generation(metadata.generation)

4. Wenn Sie den Lesevorgang auf den Bytebereich des Stripes beschränken möchten, rufen Sie set_read_range mit ReadRange::segment(offset, limit) im Request Builder auf:

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

5. Streamen Sie Chunks vom Reader und schreiben Sie jeden Chunk in die Zieldatei unter current_pos in tokio::task::spawn_blocking. Da die Standardbibliothek von Rust keine plattformübergreifende Methode für positionales Schreiben hat, wird in diesem Beispiel FileExt::write_all_at unter Unix und FileExt::seek_write unter Windows verwendet:

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

Hinweise zur Leistung

Der effektive Durchsatz von Striping-Downloads wird von mehreren Faktoren bestimmt:

• Geschwindigkeit des lokalen Laufwerks. Das Schreiben von Blöcken in einen langsamen lokalen Speicher kann zum Engpass werden. Durch die Verwendung von lokalem SSD-Speicher mit hohem Durchsatz wird sichergestellt, dass die Laufwerks-E/A-Leistung die Netzwerkleistung beim Lesen übersteigt.
• VM-CPU-Kapazität. Cloud Storage verschlüsselt Objektdaten während der Übertragung, sodass der Client jeden empfangenen Block entschlüsselt. Ihre VM muss genügend vCPUs haben, um gleichzeitige TLS-Entschlüsselungen über Stripes hinweg zu verarbeiten.
• Bucket- und VM-Standort. Das Herunterladen über Regionen hinweg führt zu WAN-Latenz und schränkt den Durchsatz ein. Platzieren Sie Ihre VM nach Möglichkeit in derselben Region wie Ihren Bucket.

Sie können auch mit der Abstimmung von stripe_size experimentieren, um zu vermeiden, dass ein sehr kleiner Reststreifen übrig bleibt oder zu viele kleine Bereichsanfragen erstellt werden.

Vollständiger Code

Im folgenden Beispiel werden die Funktionen seed, download und write_stripe zu einem vollständigen Modul kombiniert:

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

Wenn Sie download für ein 32-MiB-Objekt mit einem 1-MiB-stripe_size ausführen, sieht die Ausgabe in etwa so aus und enthält die heruntergeladene Größe, die verstrichene Zeit, die Anzahl der Stripes und die effektive Bandbreite:

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

Bereinigen

Damit Ihrem Google Cloud -Konto die von seed erstellten Testobjekte nicht in Rechnung gestellt werden, löschen Sie die Testobjekte und löschen Sie Ihren Test-Bucket, wenn Sie fertig sind. Entfernen Sie auch die lokale Zieldatei.