O Google usa tecnologia de IA na tradução de conteúdos para seu idioma de preferência. As traduções com IA podem ter erros.

Acelerar os downloads de objetos do Cloud Storage

O download sequencial de um objeto grande no Cloud Storage em uma única conexão pode deixar a largura de banda de rede disponível sem uso. Ao dividir um objeto em segmentos de intervalo de bytes chamados de faixas e ler esses segmentos simultaneamente, é possível melhorar a capacidade de download de objetos grandes usando as bibliotecas de cliente do Cloud para Rust.

Neste documento, mostramos como propagar objetos de teste grandes usando a composição de objetos, fazer o download de intervalos de bytes simultaneamente com ReadRange::segment e gravar faixas em um arquivo local em deslocamentos fixos. Se você não conhece a biblioteca de cliente, consulte Como usar o Cloud Storage com Rust primeiro.

Antes de começar

Antes de executar o código neste documento, conclua os seguintes pré-requisitos:

Crie ou selecione um projeto doGoogle Cloud com o faturamento ativado.
Crie um bucket do Cloud Storage dedicado de teste no seu projeto.
Configure o Application Default Credentials para seu ambiente.
Conceda ao principal um papel do Identity and Access Management (IAM) com permissão para ler, gravar, compor e excluir objetos no bucket, como Administrador do Storage (roles/storage.admin).

Dependências

Adicione a biblioteca de cliente do Cloud Storage e os crates de suporte ao arquivo Cargo.toml:

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

Criar objetos de teste grandes

Para testar downloads distribuídos sem fazer upload de gigabytes de dados pela rede, faça upload de um pequeno objeto de semente e use a operação de composição do Cloud Storage para concatenar cópias do lado do servidor. Em uma função assíncrona seed que aceita client, control e bucket_name no formato projects/_/buckets/BUCKET_ID, siga estas etapas:

Para referenciar objetos em solicitações de composição, importe SourceObject:
```
use google_cloud_storage::model::compose_object_request::SourceObject;
```

Faça upload de um objeto de sugestão de 1 MiB chamado 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
);

Para criar um objeto de 32 MiB chamado 32MiB.txt, concatene 32 cópias de 1MiB.txt no 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)
);

Para criar um objeto de 1 GiB chamado 1GiB.txt, concatene 32 cópias 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)
);

Para criar objetos de teste de potência de dois de 1GiB.txt (2GiB.txt, 4GiB.txt, 8GiB.txt, 16GiB.txt e 32GiB.txt), repita a operação de composição:

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

Coordenar downloads simultâneos de faixas

Para fazer o download de um objeto em paralelo, consulte o tamanho e a geração dele, pré-aloque o arquivo de destino local e gere um futuro por faixa. Em uma função download que aceita client, control, bucket_name, object_name, stripe_size e destination, siga estas etapas:

Busque os metadados do objeto usando StorageControl::get_object para inspecionar o tamanho total do objeto e bloquear as leituras subsequentes em uma única geração:
```
let metadata = control
    .get_object()
    .set_bucket(bucket_name)
    .set_object(object_name)
    .send()
    .await?;
```
Crie o arquivo de destino e pré-aloque-o para o tamanho exato do objeto para que as tarefas de worker possam gravar em deslocamentos de bytes independentes:
```
let file = Arc::new(std::fs::File::create(destination)?);
file.set_len(metadata.size as u64)?;
```

Calcule cada deslocamento de faixa usando stripe_size como limit e crie um futuro write_stripe por segmento, adicionando uma faixa parcial final quando o tamanho do objeto não for divisível 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,
    ))
}

Execute todos os futuros de stripe simultaneamente com futures::future::join_all e propague os erros:

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

Fazer o download e gravar uma faixa individual

Cada tarefa write_stripe lê um único intervalo de bytes do Cloud Storage e grava os blocos recebidos no deslocamento correspondente no identificador de processamento de arquivos local compartilhado. Para implementar write_stripe, siga estas etapas:

Importe ReadRange e defina a função 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;

Crie um builder de solicitações ReadObject com Storage::read_object e envie a solicitação depois de configurar as opções de geração e intervalo de leitura nas etapas a seguir:
```
let mut reader = client
    .read_object(&metadata.bucket, &metadata.name)
    .send()
    .await?;
```
Para fixar todas as listras na mesma geração de objeto e evitar leituras inconsistentes se outro processo substituir o objeto, chame set_generation no criador de solicitações antes de chamar send:
```
.set_generation(metadata.generation)
```
Para restringir a leitura ao intervalo de bytes da faixa, chame set_read_range com ReadRange::segment(offset, limit) no builder de solicitações:
```
.set_read_range(ReadRange::segment(offset, limit))
```

Transmita blocos do leitor e grave cada bloco no arquivo de destino em current_pos dentro de tokio::task::spawn_blocking. Como a biblioteca padrão do Rust não tem um método de gravação posicional multiplataforma, este exemplo usa FileExt::write_all_at no Unix e FileExt::seek_write no 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;
}

Considerações sobre desempenho

Vários fatores determinam a capacidade de processamento efetiva de downloads distribuídos:

Velocidade do disco local. A gravação de blocos em um armazenamento local lento pode se tornar o gargalo. O uso do armazenamento SSD local de alta capacidade de processamento ajuda a manter a E/S de disco à frente das leituras de rede.
Capacidade de CPU da VM. O Cloud Storage criptografa os dados do objeto em trânsito, então o cliente descriptografa cada bloco recebido. Verifique se a VM tem vCPUs suficientes para lidar com a descriptografia simultânea de TLS em todas as faixas.
Local do bucket e da VM. O download entre regiões introduz latência de rede de longa distância e limita a capacidade de processamento. Coloque a VM na mesma região do bucket, quando possível.

Você também pode ajustar stripe_size para evitar deixar uma faixa restante muito pequena ou criar muitas solicitações de intervalo pequenas.

Preencher código

O exemplo a seguir combina as funções seed, download e write_stripe em um 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(())
}

Quando você executa download em um objeto de 32 MiB com um stripe_size de 1 MiB, a saída é semelhante à seguinte, informando o tamanho do download, o tempo decorrido, a contagem de listras e a largura de banda efetiva:

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

Limpar

Para evitar cobranças na sua conta do Google Cloud pelos objetos de teste criados pelo seed, exclua os objetos de teste e exclua o bucket de teste quando terminar. Remova também o arquivo de destino local.