Acelera las descargas de objetos de Cloud Storage

Descargar un objeto grande en Cloud Storage de forma secuencial a través de una sola conexión puede dejar sin usar el ancho de banda de red disponible. Si divides un objeto en segmentos de rango de bytes llamados bandas y lees esos segmentos de forma simultánea, puedes mejorar la capacidad de procesamiento de descarga de objetos grandes con las bibliotecas cliente de Cloud para Rust.

En este documento, se muestra cómo propagar objetos de prueba grandes con la composición de objetos, descargar rangos de bytes de forma simultánea con ReadRange::segment y escribir bandas en un archivo local con compensaciones fijas. Si es la primera vez que usas la biblioteca cliente, consulta Usa Cloud Storage con Rust primero.

Antes de comenzar

Antes de ejecutar el código de este documento, completa los siguientes requisitos previos:

• Crea o selecciona un proyecto deGoogle Cloud con la facturación habilitada.
• Crea un bucket de Cloud Storage de prueba dedicado en tu proyecto.
• Configura las credenciales predeterminadas de la aplicación para tu entorno.
• Otorga a tu principal un rol de Identity and Access Management (IAM) con permiso para leer, escribir, redactar y borrar objetos en el bucket, como Administrador de almacenamiento (roles/storage.admin).

Nota: Cuando ejecutas el paso de propagación completo, se crean varios objetos de prueba de hasta 32 GiB cada uno (más de 63 GiB en total). Borra los objetos de prueba y el archivo de salida local en cuanto termines de realizar las pruebas para evitar cargos continuos por almacenamiento.

Dependencias

Agrega la biblioteca cliente de Cloud Storage y los crates de compatibilidad a tu archivo Cargo.toml:

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

Crea objetos de prueba grandes

Para probar las descargas divididas sin subir gigabytes de datos a través de la red, sube un objeto de origen pequeño y usa la operación de composición de Cloud Storage para concatenar copias del lado del servidor. Dentro de una función asíncrona seed que acepte client, control y bucket_name en el formato projects/_/buckets/BUCKET_ID, completa los siguientes pasos:

1. Para hacer referencia a objetos existentes en solicitudes de composición, importa SourceObject:

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

2. Sube un objeto de origen de 1 MiB llamado 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. Para crear un objeto de 32 MiB llamado 32MiB.txt, concatena 32 copias de 1MiB.txt en el servidor:

   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. Para crear un objeto de 1 GiB llamado 1GiB.txt, concatena 32 copias de 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. Para compilar objetos de prueba de potencia de dos a partir de 1GiB.txt (2GiB.txt, 4GiB.txt, 8GiB.txt, 16GiB.txt y 32GiB.txt), repite la operación de composición:

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

Coordina las descargas de bandas simultáneas

Para descargar un objeto en paralelo, consulta su tamaño y generación, asigna previamente el archivo de destino local y genera un futuro por segmento. Dentro de una función download que acepte client, control, bucket_name, object_name, stripe_size y destination, completa los siguientes pasos:

1. Recupera los metadatos del objeto con StorageControl::get_object para que puedas inspeccionar el tamaño total del objeto y bloquear las lecturas posteriores en una sola generación:

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

2. Crea el archivo de destino y asígnalo previamente al tamaño exacto del objeto para que las tareas de los trabajadores puedan escribir en compensaciones de bytes independientes:

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

3. Calcula cada desplazamiento de banda con stripe_size como limit y crea un futuro write_stripe por segmento, agregando una banda parcial final cuando el tamaño del objeto no sea divisible de manera uniforme por 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. Ejecuta todos los futuros de Stripe de forma simultánea con futures::future::join_all y propaga los errores:

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

Descarga y escribe una banda individual

Cada tarea de write_stripe lee un solo rango de bytes de Cloud Storage y escribe los fragmentos recibidos en el desplazamiento correspondiente en el manejo de archivos local compartido. Para implementar write_stripe, completa los siguientes pasos:

1. Importa ReadRange y define la función 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 compilador de solicitudes ReadObject con Storage::read_object y envía la solicitud después de configurar las opciones de generación y rango de lectura en los siguientes pasos:

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

3. Para fijar todas las bandas a la misma generación de objetos y evitar lecturas incoherentes si otro proceso reemplaza el objeto, llama a set_generation en el compilador de solicitudes antes de llamar a send:

   .set_generation(metadata.generation)

4. Para restringir la lectura al rango de bytes de la banda, llama a set_read_range con ReadRange::segment(offset, limit) en el compilador de solicitudes:

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

5. Transmite fragmentos desde el lector y escribe cada fragmento en el archivo de destino en current_pos dentro de tokio::task::spawn_blocking. Debido a que la biblioteca estándar de Rust carece de un método de escritura posicional multiplataforma, en este ejemplo se usa FileExt::write_all_at en Unix y FileExt::seek_write en 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;
   }

Consideraciones de rendimiento

Varios factores determinan la capacidad de procesamiento efectiva de las descargas divididas:

• Velocidad del disco local. Escribir fragmentos en un almacenamiento local lento puede convertirse en el cuello de botella. El uso de almacenamiento en SSD locales de alta capacidad de procesamiento ayuda a mantener la E/S de disco por delante de las lecturas de red.
• Capacidad de CPU de la VM. Cloud Storage encripta los datos de los objetos en tránsito, por lo que el cliente desencripta cada fragmento que recibe. Asegúrate de que tu VM tenga suficientes vCPUs para controlar el desencriptado de TLS simultáneo en las bandas.
• Ubicación del bucket y de la VM. La descarga entre regiones genera latencia de red de área extensa y limita la capacidad de procesamiento. Coloca tu VM en la misma región que tu bucket cuando sea posible.

También puedes experimentar con el ajuste de stripe_size para evitar dejar una franja restante muy pequeña o crear demasiadas solicitudes de rango pequeñas.

Completar código

En el siguiente ejemplo, se combinan las funciones seed, download y write_stripe en un módulo 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(())
}

Cuando ejecutas download en un objeto de 32 MiB con un stripe_size de 1 MiB, el resultado es similar al siguiente, en el que se informa el tamaño descargado, el tiempo transcurrido, el recuento de bandas y el ancho de banda efectivo:

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

Realiza una limpieza

Para evitar que se apliquen cargos a tu cuenta de Google Cloud por los objetos de prueba creados por seed, borra los objetos de prueba y borra el bucket de prueba cuando termines. También quita el archivo de destino local.