Accélérer les téléchargements d'objets Cloud Storage

Le téléchargement séquentiel d'un objet volumineux dans Cloud Storage via une seule connexion peut laisser la bande passante réseau disponible inutilisée. En divisant un objet en segments de plage d'octets appelés "bandes" et en lisant ces segments simultanément, vous pouvez améliorer le débit de téléchargement des objets volumineux à l'aide des bibliothèques clientes Cloud pour Rust.

Ce document explique comment amorcer des objets de test volumineux à l'aide de la composition d'objets, télécharger des plages d'octets simultanément avec ReadRange::segment et écrire des bandes dans un fichier local à des décalages fixes. Si vous débutez avec la bibliothèque cliente, consultez d'abord Utiliser Cloud Storage avec Rust.

Avant de commencer

Avant d'exécuter le code de ce document, assurez-vous de remplir les conditions préalables suivantes :

• Créez ou sélectionnez un projetGoogle Cloud pour lequel la facturation est activée.
• Créez un bucket Cloud Storage de test dédié dans votre projet.
• Configurez les identifiants par défaut de l'application pour votre environnement.
• Attribuez à votre compte principal un rôle IAM (Identity and Access Management) avec l'autorisation de lire, d'écrire, de composer et de supprimer des objets dans le bucket, par exemple "Administrateur de l'espace de stockage" (roles/storage.admin).

Remarque : L'exécution de l'étape d'amorçage complet crée plusieurs objets de test d'une taille maximale de 32 Gio chacun (plus de 63 Gio au total). Supprimez les objets de test et le fichier de sortie local dès que vous avez terminé les tests pour éviter des frais de stockage récurrents.

Dépendances

Ajoutez la bibliothèque cliente Cloud Storage et les crates compatibles à votre fichier Cargo.toml :

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

Créer des objets de test volumineux

Pour tester les téléchargements segmentés sans importer des gigaoctets de données sur le réseau, importez un petit objet d'amorçage et utilisez l'opération de composition de Cloud Storage pour concaténer les copies côté serveur. Dans une fonction seed asynchrone qui accepte client, control et bucket_name au format projects/_/buckets/BUCKET_ID, procédez comme suit :

1. Pour référencer des objets existants dans des requêtes de composition, importez SourceObject :

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

2. Importez un objet d'amorçage de 1 Mio nommé 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. Pour créer un objet de 32 Mio nommé 32MiB.txt, concaténez 32 copies de 1MiB.txt sur le serveur :

   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. Pour créer un objet de 1 Gio nommé 1GiB.txt, concaténez 32 copies 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. Pour créer des objets de test de puissance de deux à partir de 1GiB.txt (2GiB.txt, 4GiB.txt, 8GiB.txt, 16GiB.txt et 32GiB.txt), répétez l'opération de composition :

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

Coordonner les téléchargements de bandes simultanés

Pour télécharger un objet en parallèle, interrogez sa taille et sa génération, préallouez le fichier de destination local et générez un futur par bande. Dans une fonction download qui accepte client, control, bucket_name, object_name, stripe_size et destination, procédez comme suit :

1. Récupérez les métadonnées de l'objet à l'aide de StorageControl::get_object afin de pouvoir inspecter la taille totale de l'objet et verrouiller les lectures ultérieures sur une seule génération :

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

2. Créez le fichier de destination et préallouez-le à la taille exacte de l'objet afin que les tâches de nœud de calcul puissent écrire à des décalages d'octets indépendants :

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

3. Calculez chaque décalage de bande en utilisant stripe_size comme limit et créez un futur write_stripe par segment, en ajoutant une bande partielle finale lorsque la taille de l'objet n'est pas divisible de manière égale par 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. Exécutez tous les objets Future de bande simultanément avec futures::future::join_all et propagez les erreurs :

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

Télécharger et écrire une bande individuelle

Chaque tâche write_stripe lit une seule plage d'octets à partir de Cloud Storage et écrit les fragments reçus au décalage correspondant dans le handle de fichier local partagé. Pour implémenter write_stripe, procédez comme suit :

1. Importez ReadRange et définissez la fonction 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. Créez un générateur de requêtes ReadObject avec Storage::read_object, puis envoyez la requête après avoir configuré les options de génération et de plage de lecture en procédant comme suit :

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

3. Pour épingler toutes les bandes à la même génération d'objet et éviter les lectures incohérentes si un autre processus écrase l'objet, appelez set_generation sur le compilateur de requêtes avant d'appeler send :

   .set_generation(metadata.generation)

4. Pour limiter la lecture à la plage d'octets de la bande, appelez set_read_range avec ReadRange::segment(offset, limit) sur le compilateur de requêtes :

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

5. Diffusez des fragments à partir du lecteur et écrivez chaque fragment dans le fichier de destination à l'emplacement current_pos dans tokio::task::spawn_blocking. Étant donné que la bibliothèque standard de Rust ne dispose pas d'une méthode d'écriture positionnelle multiplate-forme, cet exemple utilise FileExt::write_all_at sous Unix et FileExt::seek_write sous 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;
   }

Considérations sur les performances

Plusieurs facteurs déterminent le débit effectif des téléchargements segmentés :

• Vitesse du disque local. L'écriture de fragments sur un espace de stockage local lent peut devenir un goulot d'étranglement. L'utilisation d'un stockage SSD local à haut débit permet de maintenir les E/S de disque en avance sur les lectures réseau.
• Capacité de processeur de la VM. Cloud Storage chiffre les données d'objet en transit. Le client déchiffre donc chaque fragment qu'il reçoit. Assurez-vous que votre VM dispose de suffisamment de processeurs virtuels pour gérer le déchiffrement TLS simultané sur les bandes.
• Emplacement du bucket et de la VM. Le téléchargement entre régions entraîne une latence du réseau étendu et limite le débit. Colocalisez votre VM dans la même région que votre bucket lorsque cela est possible.

Vous pouvez également essayer d'ajuster stripe_size pour éviter de laisser une bande restante très petite ou de créer trop de petites requêtes de plage.

Compléter le code

L'exemple suivant combine les fonctions seed, download et write_stripe dans un module complet :

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

Lorsque vous exécutez download sur un objet de 32 Mio avec un stripe_size de 1 Mio, le résultat ressemble à ce qui suit, indiquant la taille téléchargée, le temps écoulé, le nombre de bandes et la bande passante effective :

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

Effectuer un nettoyage

Pour éviter que les objets de test créés par seed ne soient facturés sur votre compte Google Cloud , supprimez les objets de test et supprimez votre bucket de test lorsque vous avez terminé. Supprimez également le fichier de destination local.