Velocizza i download degli oggetti Cloud Storage

Il download sequenziale di un oggetto di grandi dimensioni in Cloud Storage su una singola connessione può lasciare inutilizzata la larghezza di banda di rete disponibile. Dividendo un oggetto in segmenti di intervalli di byte chiamati stripe e leggendo questi segmenti contemporaneamente, puoi migliorare la velocità effettiva di download per oggetti di grandi dimensioni utilizzando le librerie client di Cloud per Rust.

Questo documento mostra come eseguire il seeding di oggetti di test di grandi dimensioni utilizzando la composizione degli oggetti, scaricare intervalli di byte contemporaneamente con ReadRange::segment e scrivere stripe in un file locale con offset fissi. Se non hai mai utilizzato la libreria client, consulta prima Utilizzo di Cloud Storage con Rust.

Prima di iniziare

Prima di eseguire il codice in questo documento, completa i seguenti prerequisiti:

• Crea o seleziona un progettoGoogle Cloud con la fatturazione abilitata.
• Crea un bucket Cloud Storage di test dedicato nel tuo progetto.
• Configura le Credenziali predefinite dell'applicazione per il tuo ambiente.
• Concedi alla tua entità un ruolo Identity and Access Management (IAM) con l'autorizzazione per leggere, scrivere, comporre ed eliminare oggetti nel bucket, ad esempio Storage Admin (roles/storage.admin).

Nota: l'esecuzione del passaggio di seeding completo crea più oggetti di test fino a 32 GiB ciascuno (oltre 63 GiB in totale). Elimina gli oggetti di test e il file di output locale non appena hai terminato i test per evitare addebiti continui per lo spazio di archiviazione.

Dipendenze

Aggiungi la libreria client di Cloud Storage e i crate di supporto al file Cargo.toml:

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

Crea oggetti di test di grandi dimensioni

Per testare i download in parallelo senza caricare gigabyte di dati sulla rete, carica un piccolo oggetto seed e utilizza l'operazione di composizione di Cloud Storage per concatenare le copie lato server. All'interno di una funzione seed asincrona che accetta client, control e bucket_name nel formato projects/_/buckets/BUCKET_ID, completa i seguenti passaggi:

1. Per fare riferimento a oggetti esistenti nelle richieste di composizione, importa SourceObject:

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

2. Carica un oggetto seed da 1 MiB denominato 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. Per creare un oggetto da 32 MiB denominato 32MiB.txt, concatena 32 copie di 1MiB.txt sul 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. Per creare un oggetto da 1 GiB denominato 1GiB.txt, concatena 32 copie di 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. Per creare oggetti di test con dimensioni pari a una potenza di due da 1GiB.txt (2GiB.txt, 4GiB.txt, 8GiB.txt, 16GiB.txt e 32GiB.txt), ripeti l'operazione di composizione:

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

Coordinare i download simultanei di stripe

Per scaricare un oggetto in parallelo, esegui una query sulle sue dimensioni e sulla sua generazione, prealloca il file di destinazione locale e genera un future per stripe. All'interno di una funzione download che accetta client, control, bucket_name, object_name, stripe_size e destination, completa i seguenti passaggi:

1. Recupera i metadati dell'oggetto utilizzando StorageControl::get_object in modo da poter esaminare le dimensioni totali dell'oggetto e bloccare le letture successive a una singola generazione:

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

2. Crea il file di destinazione e preassegnalo alle dimensioni esatte dell'oggetto in modo che le attività dei worker possano scrivere in offset di byte indipendenti:

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

3. Calcola l'offset di ogni stripe utilizzando stripe_size come limit e crea un write_stripe future per segmento, aggiungendo uno stripe parziale finale quando la dimensione dell'oggetto non è divisibile in modo uniforme per 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. Esegui tutti i future di stripe contemporaneamente con futures::future::join_all e propaga eventuali errori:

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

Scarica e scrivi una singola striscia

Ogni attività write_stripe legge un singolo intervallo di byte da Cloud Storage e scrive i blocchi ricevuti nell'offset corrispondente nell'handle di gestione di file locale condiviso. Per implementare write_stripe, completa i seguenti passaggi:

1. Importa ReadRange e definisci la funzione 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. Crea un generatore di richieste ReadObject con Storage::read_object e invia la richiesta dopo aver configurato le opzioni di generazione e intervallo di lettura nei seguenti passaggi:

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

3. Per bloccare tutti gli stripe alla stessa generazione di oggetti ed evitare letture incoerenti se un altro processo sovrascrive l'oggetto, chiama set_generation sul generatore di richieste prima di chiamare send:

   .set_generation(metadata.generation)

4. Per limitare la lettura all'intervallo di byte dello stripe, chiama set_read_range con ReadRange::segment(offset, limit) sul generatore di richieste:

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

5. Trasmetti in streaming i blocchi dal lettore e scrivi ogni blocco nel file di destinazione in current_pos all'interno di tokio::task::spawn_blocking. Poiché la libreria standard di Rust non dispone di un metodo di scrittura posizionale multipiattaforma, questo esempio utilizza FileExt::write_all_at su Unix e FileExt::seek_write su 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;
   }

Considerazioni sulle prestazioni

Diversi fattori determinano il throughput effettivo dei download con striping:

• Velocità del disco locale. La scrittura di blocchi in uno spazio di archiviazione locale lento può diventare il collo di bottiglia. L'utilizzo di uno spazio di archiviazione SSD locale con throughput elevato aiuta a mantenere l'I/O del disco in vantaggio rispetto alle letture di rete.
• Capacità della CPU della VM. Cloud Storage cripta i dati degli oggetti in transito, quindi il client decripta ogni blocco che riceve. Assicurati che la VM disponga di vCPU sufficienti per gestire la decrittografia TLS simultanea tra gli stripe.
• Posizione del bucket e della VM. Il download tra regioni introduce la latenza della wide area network e limita la velocità effettiva. Se possibile, colloca la VM nella stessa regione del bucket.

Puoi anche sperimentare con l'ottimizzazione di stripe_size per evitare di lasciare una striscia rimanente molto piccola o di creare troppe richieste di intervallo di piccole dimensioni.

Completamento codice

L'esempio seguente combina le funzioni seed, download e write_stripe in un modulo completo:

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

Quando esegui download su un oggetto da 32 MiB con un stripe_size da 1 MiB, l'output è simile al seguente e riporta le dimensioni del download, il tempo trascorso, il numero di stripe e la larghezza di banda effettiva:

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

Esegui la pulizia

Per evitare che al tuo account Google Cloud vengano addebitati costi per gli oggetti di test creati da seed, elimina gli oggetti di test ed elimina il bucket di test al termine. Rimuovi anche il file di destinazione locale.