Simultaneidade no AWS SDK para Rust

O AWS SDK para Rust não fornece controle de simultaneidade, mas os usuários têm muitas opções para implementar seus próprios controles.

Termos

Os termos relacionados a esse assunto são fáceis de confundir, e alguns deles até se tornaram sinônimos, embora originalmente representassem conceitos diferentes. Neste guia, definiremos o seguinte:

Um exemplo simples

Nosso primeiro exemplo é um programa sequencial. Em exemplos posteriores, alteraremos esse código usando técnicas de simultaneidade. Exemplos posteriores reutilizam o mesmo método build_client_and_list_objects_to_download() e fazem alterações em main(). Execute os seguintes comandos para adicionar dependências ao seu projeto:

O seguinte exemplo de tarefa é para baixar todos os arquivos em um bucket do Amazon Simple Storage Service:

use aws_sdk_s3::{Client, Error};
const EXAMPLE_BUCKET: &str = "amzn-s3-demo-bucket";  // Update to name of bucket you own.

// This initialization function won't be reproduced in
// examples following this one, in order to save space.
async fn build_client_and_list_objects_to_download() -> (Client, Vec<String>) {
    let cfg = aws_config::load_defaults(aws_config::BehaviorVersion::latest()).await;
    let client = Client::new(&cfg);
    let objects_to_download: Vec<_> = client
        .list_objects_v2()
        .bucket(EXAMPLE_BUCKET)
        .send()
        .await
        .expect("listing objects succeeds")
        .contents()
        .into_iter()
        .flat_map(aws_sdk_s3::types::Object::key)
        .map(ToString::to_string)
        .collect();
         
    (client, objects_to_download)
}

#[tokio::main]
async fn main() {
    let (client, objects_to_download) =
        build_client_and_list_objects_to_download().await;
    
    for object in objects_to_download {
        let res = client
            .get_object()
            .key(&object)
            .bucket(EXAMPLE_BUCKET)
            .send()
            .await
            .expect("get_object succeeds");
        let body = res.body.collect().await.expect("reading body succeeds").into_bytes();
        std::fs::write(object, body).expect("write succeeds");
    }
}

Devido à arquitetura dos computadores modernos, podemos reescrever esse programa para ser muito mais eficiente. Faremos isso em um exemplo posterior, mas primeiro, vamos aprender mais alguns conceitos.

Propriedade e mutabilidade

Cada valor no Rust tem um único proprietário. Quando um proprietário sai do escopo, todos os valores que ele possui também são descartados. O proprietário pode fornecer uma ou mais referências imutáveis a um valor ou uma única referência mutável. O compilador do Rust é responsável por garantir que nenhuma referência ultrapasse o limite do proprietário.

Planejamento e design adicionais são necessários quando várias tarefas precisam acessar mutavelmente o mesmo recurso. Na computação sequencial, cada tarefa pode acessar mutavelmente o mesmo recurso sem contenção porque elas são executadas uma após a outra em sequência. No entanto, na computação simultânea, as tarefas podem ser executadas em qualquer ordem e ao mesmo tempo. Portanto, devemos fazer mais para provar ao compilador que não é possível executar várias referências mutáveis (ou pelo menos travar se ocorrerem).

A biblioteca padrão do Rust fornece muitas ferramentas para ajudar. Para obter mais informações sobre esses tópicos, consulte Variables and Mutability e Understanding Ownership no livro da linguagem de programação Rust.

Mais termos!

As listas a seguir são de “objetos de sincronização”. Juntas, elas são as ferramentas necessárias para convencer o compilador de que nosso programa simultâneo não violará as regras de propriedade.

Objetos de sincronização de biblioteca padrão:

Objetos de sincronização do Tokio:

Embora os AWS SDKs tenham a intenção de ser async independentes de runtime, recomendamos usar objetos de sincronização do tokio para casos específicos.

Como reescrever o exemplo para ser mais eficiente (simultaneidade de thread único)

No exemplo modificado a seguir, usamos futures_util::future::join_all para executar TODAS as solicitações get_object simultaneamente. Execute o seguinte comando para adicionar uma nova dependência ao projeto:

#[tokio::main]
async fn main() {
    let (client, objects_to_download) =
        build_client_and_list_objects_to_download().await;
        
    let get_object_futures = objects_to_download.into_iter().map(|object| {
        let req = client
            .get_object()
            .key(&object)
            .bucket(EXAMPLE_BUCKET);

        async {
            let res = req
                .send()
                .await
                .expect("get_object succeeds");
            let body = res.body.collect().await.expect("body succeeds").into_bytes();
           // Note that we MUST use the async runtime's preferred way
           // of writing files. Otherwise, this call would block,
           // potentially causing a deadlock.
            tokio::fs::write(object, body).await.expect("write succeeds");
        }
    });

    futures_util::future::join_all(get_object_futures).await;
}

Essa é a maneira mais simples de se beneficiar da simultaneidade, mas também tem alguns problemas que podem não ser óbvios à primeira vista:

Para corrigir esses problemas, precisamos limitar a quantidade de solicitações que estamos enviando a qualquer momento. Faremos isso com um semáforo tokio:

use std::sync::Arc;
use tokio::sync::Semaphore;
const CONCURRENCY_LIMIT: usize = 50; 

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let (client, objects_to_download) =
        build_client_and_list_objects_to_download().await;
    let concurrency_semaphore = Arc::new(Semaphore::new(CONCURRENCY_LIMIT));

    let get_object_futures = objects_to_download.into_iter().map(|object| {
        // Since each future needs to acquire a permit, we need to clone
        // the Arc'd semaphore before passing it in.
        let semaphore = concurrency_semaphore.clone();
        // We also need to clone the client so each task has its own handle.
        let client = client.clone();
        async move {
            let permit = semaphore
                .acquire()
                .await
                .expect("we'll get a permit if we wait long enough");
            let res = client
                .get_object()
                .key(&object)
                .bucket(EXAMPLE_BUCKET)
                .send()
                .await
                .expect("get_object succeeds");
            let body = res.body.collect().await.expect("body succeeds").into_bytes();
            tokio::fs::write(object, body).await.expect("write succeeds");
            std::mem::drop(permit);
        }
    });

    futures_util::future::join_all(get_object_futures).await;
}

Corrigimos o possível problema de uso de memória movendo a criação da solicitação para o bloco async. Dessa forma, as solicitações não serão criadas até a hora de serem enviadas.

Também corrigimos o problema de enviar muitas solicitações de uma só vez, limitando as solicitações em andamento a CONCURRENCY_LIMIT.

Como reescrever o exemplo para ser mais eficiente (simultaneidade multiencadeada)

Nos dois exemplos anteriores, realizamos solicitações simultaneamente. Embora isso seja mais eficiente do que executá-los de forma síncrona, podemos tornar os processos ainda mais eficientes usando multiencadeação. Para fazer isso com tokio, precisaremos gerá-las como tarefas separadas.

Execute o seguinte comando para adicionar uma nova dependência ao projeto:

// Set this based on the amount of cores your target machine has.
const THREADS: usize = 8; 

#[tokio::main(flavor = "multi_thread")]
async fn main() {
    let (client, objects_to_download) =
        build_client_and_list_objects_to_download().await;
    let concurrency_semaphore = Arc::new(Semaphore::new(THREADS));

    let get_object_task_handles = objects_to_download.into_iter().map(|object| {
        // Since each future needs to acquire a permit, we need to clone
        // the Arc'd semaphore before passing it in.
        let semaphore = concurrency_semaphore.clone();
        // We also need to clone the client so each task has its own handle.
        let client = client.clone();
        
        // Note this difference! We're using `tokio::task::spawn` to
        // immediately begin running these requests.
        tokio::task::spawn(async move {
            let permit = semaphore
                .acquire()
                .await
                .expect("we'll get a permit if we wait long enough");
            let res = client
                .get_object()
                .key(&object)
                .bucket(EXAMPLE_BUCKET)
                .send()
                .await
                .expect("get_object succeeds");
            let body = res.body.collect().await.expect("body succeeds").into_bytes();
            tokio::fs::write(object, body).await.expect("write succeeds");
            std::mem::drop(permit);
        })
    });

    futures_util::future::join_all(get_object_task_handles).await;
}

Dividir o trabalho em tarefas pode ser complexo. Fazer E/S (entrada/saída) normalmente causa um bloqueio. Os runtimes podem ter dificuldade em equilibrar as necessidades de tarefas de longa duração com as de tarefas de curta duração. Seja qual for o runtime escolhido, leia as recomendações deles sobre a maneira mais eficiente de dividir seu trabalho em tarefas. Para obter as recomendações do runtime tokio, consulte Módulo tokio::task.

Depurar aplicações com vários segmentos

As tarefas executadas simultaneamente podem ser executadas em qualquer ordem. Dessa forma, os logs de programas simultâneos podem ser muito difíceis de ler. No SDK para Rust, recomendamos usar o sistema de registro em log tracing. Ele pode agrupar logs com suas tarefas específicas, não importa quando estejam em execução. Para obter orientações, consulte Configurar e usar o registro em log no AWS SDK para Rust.

Uma ferramenta muito útil para identificar tarefas bloqueadas é tokio-console, uma ferramenta de diagnóstico e depuração para programas Rust assíncronos. Ao instrumentar e executar seu programa e executar a aplicação tokio-console, é possível exibir uma visualização ao vivo das tarefas que o programa está executando. Essa exibição inclui informações úteis, como quanto tempo uma tarefa gastou esperando para adquirir recursos compartilhados ou a quantas vezes ela foi pesquisada.