Fundamentos da Concorrência em C++

A execução concorrente de programas em C++ pode ser abordada de duas maneiras principais: por meio de múltiplos processos ou por meio de múltiplos threads.

Concorrência Multiprocesso

Nesta abordagem, a aplicação é dividida em vários processos independentes, cada um executando simultaneamente. A comunicação entre esses processos ocorre através de mecanismos de comunicação interprocessos (IPC), como sinais, sockets, arquivos, pipes, entre outros.

Vantagens: O sistema operacional oferece proteção e mecanismos de comunicação de alto nível entre processos, facilitando a escrita de código concorrente seguro. Processos independentes podem até mesmo ser executados em máquinas diferentes através de conexões de rede.

Desvantagens: Devido à proteção do sistema operacional, a comunicação entre processos tende a ser menos eficiente. Além disso, a execução de múltiplos processos envolve sobrecargas fixas, como o tempo de inicialização e o gerenciamento de recursos pelo sistema operacional.

Concorrência Multithread

Aqui, múltiplos threads são executados dentro de um único processo. Threads são como processos "leves", capazes de operar de forma independente, seguindo diferentes sequências de instruções. Contudo, threads dentro do mesmo processo compartilham o mesmo espaço de endereçamento de memória e podem acessar a maioria dos dados em comum.

Vantagens: A ausência de proteção de dados entre threads na memória compartilhada reduz a carga de trabalho do sistema operacional, tornando a sobrecarga do multithreading consideravelmente menor do que a do multiprocessamento.

Desvantagens: O desenvolvimento multithread exige um esforço significativo para gerenciar a consistência dos dados compartilhados e a comunicação entre threads.

Este artigo foca exclusivamente na concorrência multithread.

Existem duas abordagens principais para usar a concorrência e melhorar o desempenho:

1. Dividir uma única tarefa em partes que podem ser executadas em paralelo para reduzir o tempo total de execução. Isso é conhecido como paralelismo de tarefas. É um processo complexo devido a possíveis dependências entre as partes.
2. Fazer com que cada thread execute a mesma operação em diferentes blocos de dados. Esta técnica é chamada de paralelismo de dados.

Quando Evitar a Concorrência

A única razão para não utilizar a concorrência é quando os custos superam os benefícios.

1. Programas concorrentes são frequentemente mais complexos de entender, o que acarreta custos adicionais de manutenção.
2. Os ganhos de desempenho podem ser menores que o esperado. Há uma sobrecarga inerente ao iniciar threads, pois o sistema operacional precisa alocar recursos do kernel e espaço de pilha para adicionar o novo thread ao agendador. Se o tempo de execução real for muito menor do que o tempo de inicialização do thread, o desempenho geral pode diminuir.
3. Em aplicações cliente-servidor que precisam lidar com um grande volume de conexões, a criação de um thread separado para cada conexão pode esgotar os recursos do sistema. Nesses casos, uma abordagem como um pool de threads pode otimizar o desempenho.

Gerenciamento de Threads

Todo programa C++ possui pelo menos um thread que executa a função main(). Outros threads podem ser executados concomitantemente com o thread principal. Assim como a função main() termina quando sua execução se completa, um thread termina após a conclusão da função que ele executa.

Em C++, iniciar um novo thread envolve a construção de um objeto std::thread.

void executar_tarefa_qualquer();
std::thread meu_thread(executar_tarefa_qualquer);

Objetos std::thread também podem ser construídos a partir de instâncias de classes que possuem um operador de chamada de função (functor):

class ProcessoEmSegundoPlano
{
public:
    void operator()() const
    {
        faz_algo_util();
        faz_outra_coisa_importante();
    }
};

ProcessoEmSegundoPlano tarefa_bg;
std::thread meu_thread(tarefa_bg); // O objeto thread internamente invoca o operador()

O objeto de função fornecido será copiado para o novo thread, e a execução e chamada do operador de função ocorrerão no espaço de memória do novo thread.

Após iniciar um thread, é preciso decidir se ele será aguardado até sua conclusão (joined) ou se ele será executado de forma independente (detached). Se essa decisão não for tomada antes da destruição do objeto std::thread, o programa será encerrado (o destrutor de std::thread invoca std::terminate()).

Se você não aguardar a junção de um thread, deve garantir a validade dos dados acessados pelo thread até que ele termine. Este não é um problema novo – o acesso a um objeto após sua destruição é comportamento indefinido em código single-thread – mas o ciclo de vida dos threads aumenta a probabilidade de tal problema. Considere o seguinte exemplo:

struct FuncaoComEstado
{
    int& referencia_int;
    FuncaoComEstado(int& ref_val) : referencia_int(ref_val) {}
    void operator() ()
    {
        for (unsigned k = 0; k < 1000000; ++k)
        {
            processar_dado(referencia_int); // 1 Potencial acesso a referência inválida
        }
    }
};

void funcao_principal_com_problema()
{
    int estado_local = 0;
    FuncaoComEstado minha_funcao(estado_local); // 'estado_local' é uma variável local
    std::thread novo_thread(minha_funcao);
    novo_thread.detach(); // O thread não será aguardado
}

Quando funcao_principal_com_problema() termina, o thread recém-criado pode ainda estar em execução, enquanto minha_funcao (e estado_local) já foi destruída. Isso pode levar o thread a acessar uma variável que não existe mais, resultando em comportamento indefinido.

Uma abordagem comum para evitar isso é copiar os dados para o thread. Se você usar um objeto invocável como função do thread, esse objeto será copiado para o thread, e o objeto original será destruído imediatamente. No entanto, é preciso ter cautela com ponteiros e referências contidos no objeto.

Alternativamente, você pode usar join() para garantir que o thread termine antes que a função principal seja concluída.

Para aguardar um thread, utilize join(). A chamada a join() aguarda o thread e libera os recursos associados. join() pode ser chamado apenas uma vez. Após a chamada, joinable() retornará false para aquele objeto std::thread.

Considerações sobre Exceções

Se você optar por desvincular um thread, basta chamar detach() logo após iniciá-lo. Se for necessário aguardar um thread, a posição da chamada join() deve ser cuidadosamente escolhida. Se uma exceção for lançada após o início do thread, mas antes da chamada join(), o join() pode ser ignorado, levando à terminação do programa.

Para evitar que a aplicação seja encerrada por exceções, é comum invocar join() dentro de um bloco catch, além de sua posição normal, garantindo que o thread seja aguardado mesmo em caso de erro:

struct FuncaoComEstado; // Definida acima
void funcao_segura()
{
    int estado_local = 0;
    FuncaoComEstado minha_funcao(estado_local);
    std::thread t(minha_funcao); // Inicia o thread
    try
    {
        faz_algo_no_thread_atual();
    }
    catch(...)
    {
        t.join(); // (1) Chamada em caso de exceção
        throw;
    }
    t.join(); // (2) Chamada normal
}

Para simplificar e tornar o código mais robusto, o padrão RAII (Resource Acquisition Is Initialization) pode ser aplicado. Uma classe guard pode ser criada, onde join() é chamado no destrutor.

// Utilizando RAII para aguardar threads
class GerenciadorDeThread
{
    std::thread& t_referencia;
public:
    explicit GerenciadorDeThread(std::thread& t_):
        t_referencia(t_)
    {}
    ~GerenciadorDeThread()
    {
        if(t_referencia.joinable())
        {
            t_referencia.join();
        }
    }
    GerenciadorDeThread(const GerenciadorDeThread&)=delete; // 3: Impede cópia
    GerenciadorDeThread& operator=(const GerenciadorDeThread&)=delete; // Impede atribuição
};

struct FuncaoComEstado; // Definida acima
void outra_funcao_segura()
{
    int estado_local = 0;
    FuncaoComEstado minha_funcao(estado_local);
    std::thread t(minha_funcao);
    GerenciadorDeThread guarda_thread(t); // Objeto RAII
    faz_alguma_outra_coisa();
}

Quando outra_funcao_segura() termina, os objetos locais são destruídos em ordem inversa. O objeto guarda_thread é o primeiro a ser destruído, e seu destrutor garante que o thread t seja aguardado. Isso acontece mesmo que faz_alguma_outra_coisa() lance uma exceção.

Threads em Segundo Plano (Detached)

O uso de detach() permite que um thread seja executado em segundo plano, sem que o thread principal precise interagir diretamente com ele. Uma vez que um thread é desvinculado, não há mais nenhum objeto std::thread que possa referenciá-lo, e ele não pode mais ser aguardado. A biblioteca C++ Runtime garante que os recursos associados ao thread sejam corretamente recuperados quando ele termina.

Threads desvinculados são frequentemente chamados de daemon threads. Sua vida útil pode se estender do início ao fim da aplicação. Eles são ideais para tarefas "fire and forget" (disparar e esquecer), onde a aplicação não precisa se preocupar com o resultado imediato ou o término do thread.

Para desvincular um thread, chame a função membro detach() do objeto std::thread. Depois disso, o objeto std::thread correspondente não terá mais relação com o thread de execução, e este último não poderá ser aguardado.

std::thread t(executar_trabalho_em_segundo_plano);
t.detach();
assert(!t.joinable()); // O objeto 't' não está mais associado a um thread executando

É importante notar que detach() não pode ser usado em um objeto std::thread que não está associado a um thread de execução. A verificação para isso é feita usando joinable(): detach() só deve ser chamado se joinable() retornar true.

Um exemplo simples de uso de threads desvinculados é um software de edição de texto que permite editar vários documentos simultaneamente. Um thread é alocado para cada janela de edição de documento, executando o mesmo código:

// Usando threads detached para gerenciar documentos
void editar_documento(std::string const& nome_arquivo)
{
    abrir_documento_e_exibir_gui(nome_arquivo);
    while(!documento_nao_editado())
    {
        ComandoUsuario comando_usr = obter_input_usuario();
        if(comando_usr.tipo == TipoComando::abrir_novo_documento)
        {
            std::string const novo_nome = obter_nome_arquivo_do_usuario();
            // Aloca um novo thread para a tarefa e o desvincula
            std::thread novo_doc_thread(editar_documento, novo_nome); // 1
            novo_doc_thread.detach(); // 2
        }
        else
        {
            processar_input_usuario(comando_usr);
        }
    }
}

É possível passar não apenas o nome da função para o construtor de std::thread, mas também os argumentos necessários para a função. Também é possível usar ponteiros para funções membro com seus respectivos objetos.

Passando Argumentos

É crucial entender que os argumentos são copiados para o espaço de memória do novo thread (como variáveis temporárias), mesmo que os parâmetros da função sejam referências. A operação de cópia sempre será executada.

void func_com_params(int i, std::string const& s);
std::thread meu_thread_com_params(func_com_params, 3, "olá");

Atenção! Se func_com_params espera um std::string como segundo argumento, mas um literal de string (const char*) é fornecido, a conversão para std::string ocorrerá no contexto do novo thread. Isso pode ser problemático se o ponteiro para o literal se tornar inválido antes que a conversão ocorra. Garanta que todas as conversões implícitas aconteçam antes de passar para o construtor de std::thread.

void func_segura_params(int i, std::string const& s);
void outro_exemplo_com_cuidado(int algum_param)
{
    char buffer[1024];
    obter_conteudo_buffer(buffer);
    // 1 (Potencialmente inseguro: buffer pode ser destruído antes da conversão no thread)
    // std::thread t(func_segura_params, 3, buffer); 
    // 2 (Seguro: std::string criada antes de ser passada)
    std::thread t(func_segura_params, 3, std::string(buffer)); 
}

No caso 1, a função outro_exemplo_com_cuidado() pode terminar antes que o buffer seja implicitamente convertido para std::string no novo thread, levando a comportamento indefinido. O caso 2 garante que a conversão seja concluída antes de ser passada para o construtor de std::thread.

Quando o objetivo é que o thread atualize uma estrutura de dados passada como argumento:

void atualizar_dados_para_componente(IdentificadorComponente w, DadosComponente& dados); // 1
void funcao_com_erro_novamente(IdentificadorComponente w)
{
    DadosComponente meus_dados;
    std::thread t(atualizar_dados_para_componente, w, meus_dados); // 2: 'meus_dados' será copiada!
    exibir_status();
    t.join();
    processar_dados_componente(meus_dados); // 'meus_dados' aqui ainda é a cópia original
}

Embora atualizar_dados_para_componente espere uma referência, o construtor de std::thread realiza uma cópia cega, ignorando o tipo de parâmetro. Para tipos móveis, o código interno passaria o argumento como um rvalue, mas como a função espera uma referência não-constante (e não um rvalue), isso resultaria em um erro de compilação. A solução é:

std::thread t(atualizar_dados_para_componente, w, std::ref(meus_dados));

Dessa forma, atualizar_dados_para_componente receberá uma referência para meus_dados, e não uma cópia, permitindo que o código compile e funcione conforme o esperado.

A sintaxe de passagem de parâmetros é similar à de std::bind. É também possível passar um ponteiro para uma função membro e um ponteiro para o objeto apropriado como primeiro argumento:

class MinhaClasse
{
public:
    void executar_trabalho_longo();
};

MinhaClasse objeto_x;
std::thread t_obj_membro(&MinhaClasse::executar_trabalho_longo, &objeto_x); // 1

O novo thread invocará objeto_x.executar_trabalho_longo(), com o endereço de objeto_x fornecido como ponteiro para o objeto. Argumentos adicionais para a função membro podem ser passados em seguida.

class MinhaClasseComParam
{
public:
    void executar_trabalho_longo_com_param(int);
};
MinhaClasseComParam obj_y;
int numero_param = 0;
std::thread t_obj_membro_param(&MinhaClasseComParam::executar_trabalho_longo_com_param, &obj_y, numero_param);

Transferindo Propriedade de Threads

Várias classes na biblioteca padrão C++ são proprietárias de recursos (resource-owning types), como std::ifstream, std::unique_ptr e std::thread. Estes são tipos móveis, mas não copiáveis, razão pela qual as operações de movimento foram introduzidas no C++ moderno. Veja um exemplo de transferência de propriedade de threads:

void alguma_funcao_t1();
void alguma_outra_funcao_t2();

std::thread thread_alpha(alguma_funcao_t1);     // 1: Cria novo thread _1 associado a thread_alpha
std::thread thread_beta = std::move(thread_alpha); // 2: Transfere propriedade de _1 para thread_beta; thread_alpha agora não tem thread associado
thread_alpha = std::thread(alguma_outra_funcao_t2); // 3: Cria novo thread _2, transfere propriedade para thread_alpha
std::thread thread_gama;                       // 4: Cria thread_gama sem thread associado
thread_gama = std::move(thread_beta);           // 5: Transfere propriedade de _1 (de thread_beta) para thread_gama
// thread_alpha agora associado a _2
// thread_beta agora sem thread associado
// thread_gama agora associado a _1
thread_alpha = std::move(thread_gama);         // 6: CUIDADO! thread_alpha já tem um thread associado (_2). Isso causará std::terminate().

A última operação tentou transferir a propriedade do thread \_1 (de thread\_gama) para thread\_alpha. No entanto, thread\_alpha já estava associado ao thread \_2. Não é possível "descartar" um thread atribuindo um novo valor a um std::thread que já possui um thread associado sem antes join() ou detach() o thread existente. Isso resultará em uma chamada para std::terminate(), encerrando o programa.

Instâncias de std::thread suportam passagem como parâmetro e retorno de função:

// Propriedade de thread pode ser transferida fora de funções
std::thread criar_thread_um()
{
    void funcao_externa_um();
    return std::thread(funcao_externa_um);
}

std::thread criar_thread_dois()
{
    void funcao_externa_dois(int);
    std::thread t_local(funcao_externa_dois, 42);
    return t_local;
}

// Propriedade de thread pode ser transferida dentro de funções (std::thread como parâmetro)
void aceitar_thread(std::thread t_param);
void exemplo_passagem()
{
    void funcao_para_passar();
    aceitar_thread(std::thread(funcao_para_passar));
    std::thread t_movel(funcao_para_passar);
    aceitar_thread(std::move(t_movel));
}

Determinando o Número de Threads

std::thread::hardware_concurrency() retorna o número de threads que podem ser executados simultaneamente em hardware. Em sistemas multi-core, o valor de retorno pode ser o número de núcleos da CPU. Este valor é apenas uma indicação; a função retorna 0 se não for possível determinar.

Um exemplo de uso é uma versão paralela de std::accumulate, onde o trabalho é dividido em tarefas menores para cada thread, com um número mínimo de tarefas para evitar a criação excessiva de threads. O programa lançará uma exceção se a quantidade de operações for zero ou se std::thread não conseguir iniciar um thread. A decisão sobre a quantidade de threads deve levar em conta o retorno de std::thread::hardware_concurrency() e as necessidades reais da tarefa.

Identificação de Threads

O identificador de um thread é do tipo std::thread::id. Pode ser obtido de duas formas:

1. Chamando a função membro get_id() de um objeto std::thread. Se o objeto std::thread não estiver associado a nenhum thread de execução, get_id() retornará um valor construído por padrão, que representa "nenhum thread".
2. Chamando std::this_thread::get_id() (definido no cabeçalho <thread>) no thread atual.

Se dois objetos std::thread::id forem iguais, eles representam o mesmo thread ou ambos representam "nenhum thread". Se forem diferentes, representam threads distintos ou um tem um thread e o outro não. Objetos std::thread::id podem ser usados para comparações de ordenação, e a biblioteca padrão fornece std::hash<std::thread::id>, permitindo usá-los como chaves em contêineres não ordenados.

Instâncias de std::thread::id são frequentemente usadas para verificar se um thread específico deve realizar certas operações. Por exemplo, ao dividir uma tarefa entre threads, o thread principal pode ter que realizar algumas tarefas diferentes dos outros. Antes de iniciar outros threads, pode-se usar std::this_thread::get_id() para verificar se o ID do thread atual é o mesmo que o ID do thread inicial.

std::thread::id id_thread_mestre;

void parte_essencial_do_algoritmo()
{
    if(std::this_thread::get_id() == id_thread_mestre)
    {
        executar_trabalho_do_thread_mestre();
    }
    executar_trabalho_comum();
}

Dados Compartilhados

Condições de Corrida

O termo "condição de corrida" na terminologia padrão C++ refere-se à modificação concorrente de um objeto independente. A data race é a causa do comportamento indefinido.

Aqui estão algumas maneiras de evitar condições de corrida maliciosas:

1. Utilizar mecanismos de proteção para estruturas de dados, garantindo que apenas o thread modificador possa ver o estado intermediário dos invariantes. Do ponto de vista de outros threads, a modificação ou já foi concluída ou ainda não começou.
2. Adotar programação lock-free, onde modificações em estruturas de dados e invariantes são realizadas como uma série de mudanças indivisíveis.
3. Empregar memória transacional de software (STM), tratando as atualizações de estruturas de dados de forma transacional.

A abordagem mais fundamental é o uso de mutexes.

Utilizando Mutexes

Uma instância de mutex é criada instanciando std::mutex. As funções membro lock() e unlock() são usadas para bloquear e desbloquear o mutex, respectivamente. No entanto, não é recomendável chamar diretamente essas funções. Em vez disso, é preferível usar o padrão RAII para gerenciar o mutex.

A biblioteca C++ oferece a classe de template RAII std::lock_guard. Ela bloqueia o mutex no construtor e o desbloqueia no destrutor.

// Protegendo uma lista com mutex
std::list<int> lista_protegida;
std::mutex mutex_lista;

void adicionar_a_lista(int novo_valor)
{
    std::lock_guard<std::mutex> guarda_lock(mutex_lista); // 3: Adquire o lock
    lista_protegida.push_back(novo_valor);
}

bool lista_contem(int valor_a_encontrar)
{
    std::lock_guard<std::mutex> guarda_lock(mutex_lista); // 4: Adquire o lock
    return std::find(lista_protegida.begin(), lista_protegida.end(), valor_a_encontrar) != lista_protegida.end();
}

No exemplo acima, as funções adicionar_a_lista e lista_contem acessam os dados de forma mútua. Em uma arquitetura orientada a objetos, seria mais limpo encapsular a lista e o mutex em uma classe, onde adicionar_a_lista e lista_contem seriam métodos. O mutex e os dados protegidos seriam membros privados da classe, tornando o código mais claro e fácil de entender quando o mutex está bloqueado.

Para compiladores C++17 e posteriores, é possível usar std::scoped_lock guarda_lock(mutex_lista), que funciona de forma similar.

Nunca Passe Ponteiros ou Referências Não Protegidos para Fora do Escopo de um Mutex!

As condições de corrida podem ocorrer não apenas na modificação concorrente dos mesmos dados, mas também em interações entre interfaces.

Pilha<int> minha_pilha; // Supondo uma implementação de Pilha
if(!minha_pilha.vazia()) // 1
{
    int const valor = minha_pilha.topo(); // 2
    minha_pilha.remover(); // 3
    processar_valor(valor);
}

Entre a chamada a vazia() (1) e topo() (2), outro thread pode chamar remover() e remover o último elemento, tornando a pilha vazia. Mesmo que o mutex proteja os dados internos da pilha, isso não impede a condição de corrida de interface.

Problemas de Deadlock (Impasse)

Deadlocks ocorrem quando múltiplos mutexes se aguardam mutuamente, impedindo qualquer thread de progredir. A regra mais básica para evitar deadlocks é sempre bloquear os mutexes na mesma ordem: sempre bloquear o Mutex A antes do Mutex B.

No entanto, se múltiplos mutexes protegem instâncias independentes da mesma classe, e uma operação tenta trocar dados entre duas instâncias diferentes da mesma classe, ainda pode ocorrer um deadlock.

A biblioteca padrão C++ oferece std::lock, que bloqueia múltiplos mutexes de uma só vez, sem o risco de deadlock.

class ObjetoGrande;
void trocar_conteudo(ObjetoGrande& esq, ObjetoGrande& dir);

class Recurso
{
private:
    ObjetoGrande detalhe_interno;
    std::mutex m;
public:
    Recurso(ObjetoGrande const& sd): detalhe_interno(sd){}
    
    friend void trocar_recursos(Recurso& esq, Recurso& dir)
    {
        if(&esq == &dir) // Se forem o mesmo objeto, não faz nada
            return;
        std::lock(esq.m, dir.m); // Bloqueia ambos os mutexes de forma atômica
        std::lock_guard<std::mutex> guarda_a(esq.m, std::adopt_lock); // Adota o lock existente
        std::lock_guard<std::mutex> guarda_b(dir.m, std::adopt_lock); // Adota o lock existente
        trocar_conteudo(esq.m, dir.m);
    }
};

Primeiro, std::lock() bloqueia ambos os mutexes. Em seguida, dois objetos std::lock_guard são criados com o argumento std::adopt_lock, indicando que eles devem gerenciar um mutex já bloqueado.

A partir do C++17, o template RAII std::scoped_lock, com funcionalidade equivalente a std::lock_guard<>, pode aceitar um número variável de mutexes como parâmetros do template e seus respectivos mutexes como argumentos do construtor. Ele bloqueia-os no construtor e os desbloqueia no destrutor.

void trocar_recursos_cpp17(Recurso& esq, Recurso& dir)
{
    if(&esq == &dir)
        return;
    std::scoped_lock guarda_locks(esq.m, dir.m); // Bloqueia ambos de forma segura
    trocar_conteudo(esq.detalhe_interno, dir.detalhe_interno);
}

Aqui, a dedução implícita de tipos de template do C++17 é usada, equivalente a std::scoped_lock<std::mutex, std::mutex> guarda_locks(esq.m, dir.m);.

Deadlocks geralmente são causados pelo uso inadequado de locks. Mesmo a chamada mútua de join() entre dois objetos std::thread pode levar a um deadlock. Algumas técnicas para evitar deadlocks:

• Evitar locks aninhados: Se um thread já possui um lock, evite adquirir outro. Se múltiplos locks forem necessários, use std::scoped_lock() ou std::lock().
• Evitar chamar código externo enquanto um lock é mantido: O código externo pode tentar adquirir um lock, violando a regra anterior.
• Usar uma ordem fixa para adquirir locks: Quando for estritamente necessário adquirir dois ou mais locks que não podem ser gerenciados por uma única operação std::lock, tente sempre adquiri-los na mesma ordem em todos os threads.
• Usar uma estrutura de lock hierárquica.

std::unique_lock é uma classe de template RAII mais flexível, que suporta bloqueio imediato, bloqueio atrasado, tentativa de bloqueio, tentativa de bloqueio com timeout e transferência de propriedade de mutex via semântica de movimento.

Exemplo de transferência de std::unique_lock entre escopos:

std::unique_lock<std::mutex> obter_lock_preparado()
{
    extern std::mutex meu_mutex_global; // Mutex declarado externamente
    std::unique_lock<std::mutex> lk(meu_mutex_global);
    preparar_dados_para_processamento();
    return lk; // Retorna o unique_lock
}

void processar_dados()
{
    std::unique_lock<std::mutex> lk_processo(obter_lock_preparado()); // 2: Adquire o lock retornado
    fazer_algo_com_dados_processados();
}

Protegendo a Inicialização de Dados Compartilhados

Em código single-thread, a inicialização tardia é comum:

std::shared_ptr<MeuRecurso> ponteiro_recurso;

void inicializar_se_necessario()
{
    if(!ponteiro_recurso) // 1
    {
        ponteiro_recurso.reset(new MeuRecurso);
    }
    ponteiro_recurso->executar_tarefa();
}

Em código multithread, apenas o ponto 1 precisa ser protegido para garantir que a inicialização do dado compartilhado seja segura. No entanto, o uso direto de um mutex, como abaixo, serializa threads desnecessariamente. Cada thread deve esperar pelo mutex apenas para verificar se o recurso já foi inicializado.

std::shared_ptr<MeuRecurso> ponteiro_recurso_mt;
std::mutex mutex_recurso_mt;

void inicializar_serializado()
{
    std::unique_lock<std::mutex> lk(mutex_recurso_mt); // Todos os threads serializam aqui
    if(!ponteiro_recurso_mt)
    {
        ponteiro_recurso_mt.reset(new MeuRecurso); // Apenas a inicialização precisa de proteção
    }
    lk.unlock(); // Libera o lock
    ponteiro_recurso_mt->executar_tarefa();
}

O padrão Double-Checked Locking Pattern (DCLP) é conhecido por ter condições de corrida em C++. A biblioteca padrão C++ resolve isso com std::once_flag e std::call_once, que consomem menos recursos do que bloquear um mutex e verificar explicitamente um ponteiro.

std::shared_ptr<MeuRecurso> recurso_unico;
std::once_flag flag_recurso; // 1: Flag para inicialização única

void funcao_inicializa_recurso()
{
    recurso_unico.reset(new MeuRecurso);
}

void usar_recurso_foo()
{
    std::call_once(flag_recurso, funcao_inicializa_recurso); // Garante inicialização única
    recurso_unico->executar_tarefa();
}

std::call_once também pode ser usado para inicialização tardia de membros de classe (thread-safe), passando um ponteiro para a função membro e o ponteiro da instância do membro.

class ConexaoGerenciada
{
private:
    InfoConexao detalhes_conexao;
    HandleConexao manipulador_conexao;
    std::once_flag flag_inicializacao_conexao;
   	
    void abrir_conexao_interna()
    {
        manipulador_conexao = GerenciadorConexoes.abrir(detalhes_conexao);
    }
public:
    ConexaoGerenciada(InfoConexao const& detalhes_conexao_):
        detalhes_conexao(detalhes_conexao_)
    {}

    void enviar_dados(PacoteDados const& dados) // 1
    {
        std::call_once(flag_inicializacao_conexao, &ConexaoGerenciada::abrir_conexao_interna, this); // 2
        manipulador_conexao.enviar_dados(dados);
    }
    PacoteDados receber_dados() // 3
    {
        std::call_once(flag_inicializacao_conexao, &ConexaoGerenciada::abrir_conexao_interna, this); // 2
        return manipulador_conexao.receber_dados();
    }
};

Em compiladores anteriores ao C++11, a inicialização de variáveis estáticas locais tinha condições de corrida. No C++11 e posteriores, a inicialização de variáveis estáticas locais é garantida como thread-safe, ocorrendo apenas uma vez e de forma completa antes que qualquer outro thread possa acessá-la.

class MinhaClasseSingleton;
MinhaClasseSingleton& obter_instancia_minha_classe()
{
    static MinhaClasseSingleton instancia; // Inicialização thread-safe em C++11+
    return instancia;
}

C++17 introduziu std::shared_mutex e std::shared_timed_mutex (este último também disponível em C++14). Esses mutexes são conhecidos como reader-writer locks (locks de leitura-escrita). std::shared_mutex oferece maior desempenho, enquanto std::shared_timed_mutex possui mais funcionalidades. Threads que não precisam modificar a estrutura de dados podem adquirir acesso de leitura usando std::shared_lock<std::shared_mutex>.

Exemplo de proteção de estrutura de dados com std::shared_mutex:

class EntradaDNS;
class CacheDNS
{
    std::map<std::string, EntradaDNS> entradas;
    mutable std::shared_mutex mutex_entradas; // Mutex compartilhado
public:
    // Leitura
    EntradaDNS encontrar_entrada(std::string const& dominio) const
    {
        std::shared_lock<std::shared_mutex> lk(mutex_entradas); // Lock compartilhado
        std::map<std::string, EntradaDNS>::const_iterator const it = entradas.find(dominio);
        return (it == entradas.end()) ? EntradaDNS() : it->second;
    }
    // Escrita
    void atualizar_ou_adicionar_entrada(std::string const& dominio, EntradaDNS const& detalhes_dns)
    {
        std::lock_guard<std::shared_mutex> lk(mutex_entradas); // Lock exclusivo
        entradas[dominio] = detalhes_dns;
    }
};

Quando um std::shared_lock é adquirido, outros threads que também tentam adquirir um std::shared_lock não são bloqueados. No entanto, quando um std::lock_guard (ou std::unique_lock) é adquirido, qualquer outra operação de lock (compartilhado ou exclusivo) no mesmo mutex por outros threads será bloqueada.

Locks Aninhados

É um erro um thread tentar bloquear um std::mutex que ele já possui. Tentar fazer isso leva a comportamento indefinido. Em alguns casos, um thread pode precisar adquirir um mutex várias vezes antes de liberá-lo. Para isso, a biblioteca padrão C++ oferece a classe std::recursive_mutex. Ela funciona de forma idêntica a std::mutex, exceto que permite que o mesmo thread bloqueie a mesma instância várias vezes.

Operações de Sincronização de Threads

A biblioteca padrão C++ oferece variáveis de condição (condition variables) e futures para sincronização entre threads.

Uma forma rudimentar de esperar por um evento em threads é usar std::this_thread::sleep_for() em um loop de verificação periódica:

bool flag_pronta;
std::mutex m_sinc;

void aguardar_flag()
{
    std::unique_lock<std::mutex> lk(m_sinc);
    while(!flag_pronta)
    {
        lk.unlock(); // 1: Desbloqueia o mutex
        std::this_thread::sleep_for(std::chrono::milliseconds(100)); // 2: Dorme por 100ms
        lk.lock(); // 3: Bloqueia o mutex novamente
    }
}

Neste loop, a função desbloqueia o mutex (1) antes de dormir (2) e o bloqueia novamente (3) após o período de espera. Isso permite que outros threads adquiram o lock e configurem a flag. No entanto, há uma maneira mais eficiente e preferível de esperar por eventos na biblioteca padrão C++.

O mecanismo para um thread ser acordado por outro, geralmente quando há uma nova tarefa disponível, é chamado de "variável de condição".

A biblioteca padrão C++ possui duas implementações de variáveis de condição: std::condition_variable e std::condition_variable_any. Ambas requerem um mutex para funcionar. A primeira funciona apenas com std::mutex, enquanto a segunda (com o sufixo _any) pode funcionar com qualquer mutex adequado. std::condition_variable é mais eficiente, enquanto std::condition_variable_any é mais flexível.

Exemplo de uso de std::condition_variable para esperar por dados:

std::mutex mut_prod_cons;
std::queue<BlocoDados> fila_dados; // 1: Fila compartilhada
std::condition_variable cond_dados;

void thread_preparacao_dados()
{
    while(mais_dados_para_preparar())
    {
        BlocoDados const dados = preparar_proximo_dado();
        std::lock_guard<std::mutex> lk(mut_prod_cons);
        fila_dados.push(dados); // 2: Adiciona dados à fila
        cond_dados.notify_one(); // 3: Notifica um thread de espera
    }
}

void thread_processamento_dados()
{
    while(true)
    {
        std::unique_lock<std::mutex> lk(mut_prod_cons); // 4: Bloqueia o mutex
        cond_dados.wait(lk, []{return !fila_dados.empty();}); // 5: Espera até a fila não estar vazia
        BlocoDados dados_para_processar = fila_dados.front();
        fila_dados.pop();
        lk.unlock(); // 6: Desbloqueia o mutex
        processar(dados_para_processar);
        if(e_ultimo_bloco(dados_para_processar))
            break;
    }
}

Neste exemplo, dois threads interagem com uma fila de dados (1). Quando os dados estão prontos, o thread de preparação bloqueia a fila com std::lock_guard, adiciona os dados (2) e então chama notify_one() (3) para notificar um thread em espera. O thread de processamento bloqueia o mutex com std::unique_lock (4) e, em seguida, chama wait() na variável de condição (5), passando o lock e uma expressão lambda que verifica se a fila não está vazia. Se a fila estiver vazia, wait() desbloqueia o mutex e coloca o thread em estado de bloqueio/espera até ser notificado. Quando notificado, ele rebloqueia o mutex, verifica a condição novamente e, se verdadeira, continua a execução.

Utilizando Eventos Future

Futures são como delegar uma tarefa a outro thread, permitindo que o thread principal continue com outras atividades até que o resultado da tarefa seja necessário. Se a tarefa já foi concluída, o resultado é obtido imediatamente; caso contrário, o thread bloqueia até que o resultado esteja disponível.

A biblioteca padrão C++ oferece dois tipos de futures, declarados no cabeçalho <future>: unique futures (std::future<>) e shared futures (std::shared_future<>). Eles são análogos a std::unique_ptr e std::shared_ptr. std::future pode ser associado a um único evento, enquanto std::shared_future pode ser associado a múltiplos.

Para iniciar uma tarefa assíncrona sem urgência para o resultado, use std::async. Ao contrário de std::thread, std::async retorna um objeto std::future que eventualmente conterá o resultado. Quando o valor é necessário, basta chamar a função membro get() do objeto std::future, que bloqueará o thread até que o future esteja pronto e retornará o resultado.

#include <future>
#include <iostream>

int calcular_resposta_para_pergunta_complexa();
void fazer_outras_tarefas_importantes();

int main()
{
    std::future<int> a_resposta = std::async(calcular_resposta_para_pergunta_complexa);
    fazer_outras_tarefas_importantes();
    std::cout << "A resposta é " << a_resposta.get() << std::endl;
}

A passagem de argumentos para std::async é similar à de std::thread. Pode-se passar ponteiros para funções membro com seus respectivos objetos e argumentos. Se um argumento for um rvalue, ele será movido internamente.

#include <string>
#include <future>

struct MinhaEstrutura
{
    void metodo_sem_retorno(int, std::string const&);
    std::string metodo_com_retorno(std::string const&);
};

MinhaEstrutura objeto_s;
auto f1 = std::async(&MinhaEstrutura::metodo_sem_retorno, &objeto_s, 42, "olá"); // Invoca objeto_s.metodo_sem_retorno(42, "olá")
auto f2 = std::async(&MinhaEstrutura::metodo_com_retorno, objeto_s, "adeus"); // Invoca copia_de_objeto_s.metodo_com_retorno("adeus")

Na maioria dos casos, um parâmetro adicional do tipo std::launch pode ser passado para std::async antes da chamada da função. std::launch::deferred indica que a função será executada apenas quando wait() ou get() for chamado. std::launch::async garante que a função será executada em um thread separado.

auto f6 = std::async(std::launch::async, MinhaEstrutura(), 1.2); // Executa em um novo thread
auto f7 = std::async(std::launch::deferred, alguma_funcao_com_ref, std::ref(objeto_s)); // Executa em wait() ou get()
auto f8 = std::async(std::launch::deferred | std::launch::async, alguma_funcao_com_ref, std::ref(objeto_s)); // Implementação escolhe o modo
auto f9 = std::async(alguma_funcao_com_ref, std::ref(objeto_s)); // Padrão é std::launch::deferred | std::launch::async
f7.wait(); // Invoca a função adiada

Associando Future a Tarefas

std::packaged_task<> vincula um future a uma função ou objeto invocável. Quando o objeto std::packaged_task é invocado, a função ou objeto é chamado, e seu valor de retorno é armazenado, tornando o future associado pronto. O parâmetro de template de std::packaged_task<> é uma assinatura de função. Parâmetros da assinatura podem ser implicitamente convertidos. Por exemplo, uma função que aceita um int e retorna um float pode ser usada para construir uma instância de std::packaged_task<double(double)>.

std::packaged_task é similar a std::function no sentido de que encapsula um objeto invocável, mas std::packaged_task transfere o resultado da execução do invocável para um objeto std::future.

Utilizando std::promises

std::promise<T> oferece uma maneira de definir um valor que será associado a um objeto std::future<T>. O par std::promise/std::future fornece um mecanismo onde o future pode bloquear um thread de espera, e o thread que fornece os dados pode usar o promise para definir o valor associado, tornando o future "pronto".

O objeto std::future associado a um std::promise pode ser obtido através da função membro get_future() de std::promise, de forma similar ao uso de std::packaged_task. Quando o promise é "cumprido" (usando a função membro set_value()), o future correspondente entra no estado "pronto" e pode ser usado para recuperar o valor armazenado.

Futures Armazenam Exceções Lançadas

Se uma chamada de função lança uma exceção, essa exceção é armazenada no future. O future então entra no estado "pronto", e uma chamada subsequente a get() relançará a exceção armazenada.

Para std::promise, se uma exceção deve ser armazenada em vez de um valor, set_exception() deve ser chamado em vez de set_value(). Isso é frequentemente usado dentro de um bloco catch:

extern std::promise<double> meu_promise_global; // Promise declarado externamente
try
{
    meu_promise_global.set_value(calcular_valor_complexo());
}
catch(...)
{
    meu_promise_global.set_exception(std::current_exception()); // Armazena a exceção atual
}

Uma alternativa é usar std::copy_exception(), que armazena uma nova exceção sem relançá-la:

meu_promise_global.set_exception(std::copy_exception(std::logic_error("Erro de lógica personalizada")));

Esta abordagem é mais clara do que usar blocos try/catch quando o tipo da exceção é conhecido e pode oferecer otimizações significativas ao compilador.

Utilizando Timeouts

A maneira mais simples de usar timeouts é adicionar um atraso de processamento a threads específicos. Quando um thread está ocioso, ele não consome tempo de CPU que outros threads poderiam usar. As funções std::this_thread::sleep_for() e std::this_thread::sleep_until() são usadas para isso.

Programação Funcional com Futures

Quicksort sequencial:

template<typename T>
std::list<T> quick_sort_sequencial(std::list<T> entrada)
{
    if(entrada.empty())
    {
        return entrada;
    }
    std::list<T> resultado;
    resultado.splice(resultado.begin(), entrada, entrada.begin()); // 1: Move o pivô para o resultado
    T const& pivo = *resultado.begin(); // 2: Define o pivô
    auto ponto_divisao = std::partition(entrada.begin(), entrada.end(),
                                     [&](T const& t){return t < pivo;}); // 3: Particiona a lista
    std::list<T> parte_menor;
    parte_menor.splice(parte_menor.end(), entrada, entrada.begin(),
                      ponto_divisao); // 4: Move elementos menores para 'parte_menor'
    auto nova_parte_menor(
        quick_sort_sequencial(std::move(parte_menor))); // 5: Ordena recursivamente a parte menor
    auto nova_parte_maior(
        quick_sort_sequencial(std::move(entrada))); // 6: Ordena recursivamente a parte maior
    resultado.splice(resultado.end(), nova_parte_maior); // 7: Adiciona a parte maior
    resultado.splice(resultado.begin(), nova_parte_menor); // 8: Adiciona a parte menor
    return resultado;
}

Quicksort paralelo usando futures:

template<typename T>
std::list<T> quick_sort_paralelo(std::list<T> entrada)
{
    if(entrada.empty())
    {
        return entrada;
    }
    std::list<T> resultado;
    resultado.splice(resultado.begin(), entrada, entrada.begin());
    T const& pivo = *resultado.begin();
    auto ponto_divisao = std::partition(entrada.begin(), entrada.end(),
                                     [&](T const& t){return t < pivo;});
    std::list<T> parte_inferior;
    parte_inferior.splice(parte_inferior.end(), entrada, entrada.begin(),
                      ponto_divisao);
    std::future<std::list<T> > futuro_parte_inferior( // 1: Inicia tarefa assíncrona para ordenar parte inferior
        std::async(&quick_sort_paralelo<T>, std::move(parte_inferior)));
    auto nova_parte_superior(
        quick_sort_paralelo(std::move(entrada))); // 2: Ordena parte superior no thread atual
    resultado.splice(resultado.end(), nova_parte_superior); // 3: Adiciona a parte superior ordenada
    resultado.splice(resultado.begin(), futuro_parte_inferior.get()); // 4: Obtém e adiciona a parte inferior ordenada
    return resultado;
}

Em vez de usar diretamente std::async(), pode-se criar uma função auxiliar, por exemplo, iniciar_tarefa_assincrona(), que encapsula std::packaged_task e std::thread. Embora isso por si só não ofereça grandes vantagens (e pode causar sobrecarga em larga escala), serve como base para uma implementação mais complexa, como adicionar tarefas a uma fila e executá-las usando um pool de threads. std::async é mais adequado quando todas as tarefas são conhecidas e há controle total sobre o pool de threads.

Implementação simples de iniciar_tarefa_assincrona:

template<typename F, typename A>
std::future<typename std::result_of<F(A&&)>::type>
    iniciar_tarefa_assincrona(F&& f, A&& a)
{
    typedef typename std::result_of<F(A&&)>::type tipo_resultado;
    std::packaged_task<tipo_resultado(A&&)>
        tarefa(std::move(f));
    std::future<tipo_resultado> res(tarefa.get_future());
    std::thread t(std::move(tarefa), std::move(a));
    t.detach(); // Desvincula o thread
    return res;
}

A programação funcional pode ser vista como um paradigma para programação concorrente, assim como o Communicating Sequential Processes (CSP), onde threads não compartilham dados, mas se comunicam através de canais.

Lógicas de Caixa Eletrônico (ATM) Simples Baseadas no Padrão Actor

Modelagem usando máquina de estados:

struct CartaoInserido
{
    std::string conta;
};

class CaixaEletronico
{
    sistema_mensagens::receptor entrada;
    sistema_mensagens::emissor banco;
    sistema_mensagens::emissor hardware_interface;
    void (CaixaEletronico::*estado_atual)(); // Ponteiro para função membro de estado
    std::string conta_usuario;
    std::string pin_digitado;

    void aguardando_cartao() // 1
    {
        hardware_interface.enviar(exibir_mensagem_inserir_cartao()); // 2
        entrada.esperar(). // 3
            manipular<CartaoInserido>(
            [&](CartaoInserido const& msg) // 4
            {
                conta_usuario = msg.conta;
                pin_digitado = "";
                hardware_interface.enviar(exibir_mensagem_digitar_pin());
                estado_atual = &CaixaEletronico::obtendo_pin;
            }
        );
    }
    void obtendo_pin(); // Declaração da função de estado

public:
    void executar() // 5
    {
        estado_atual = &CaixaEletronico::aguardando_cartao; // 6: Estado inicial
        try
        {
            for(;;) // Loop infinito
            {
                (this->*estado_atual)(); // 7: Executa o estado atual
            }
        }
        catch(sistema_mensagens::fila_fechada const&)
        {
            // Tratamento para fechamento da fila de mensagens
        }
    }
};

Aqui, a preocupação não é com sincronização e concorrência, mas com o recebimento e envio de mensagens. A máquina de estados da lógica do ATM é executada em um thread independente, assim como as interfaces de comunicação com o banco e o terminal. Esses "atores" enviam mensagens uns aos outros para executar suas tarefas, sem compartilhar estado diretamente, a menos que seja explicitamente passado via mensagem.

Sincronização de Eventos com Latches (std::latch)

Ao construir um std::latch, o valor do contador é passado como o único argumento do construtor. Quando um evento de espera ocorre, a função membro count_down do latch é chamada. O latch entra no estado "pronto" quando o contador atinge zero. A função membro wait() pode ser usada para bloquear o thread até que o latch esteja pronto. is_ready() verifica se o latch está pronto. Para diminuir o contador em um e bloquear até que seja zero, chame count_down_and_wait().

void exemplo_latch(){
    unsigned const num_threads_config = /* algum valor */;
    std::latch final_tarefas(num_threads_config); // 1: Latch com contador inicial
    DadosProcessados dados_partes[num_threads_config];
    std::vector<std::future<void> > lista_threads_async;

    for(unsigned i = 0; i < num_threads_config; ++i)
        lista_threads_async.push_back(std::async(std::launch::async,[&,i]{ // 2: Inicia tarefa assíncrona
            dados_partes[i] = gerar_dados_para_processamento(i);
            final_tarefas.count_down(); // 3: Decrementa o contador do latch
            fazer_mais_alguma_coisa_local(); // 4: Trabalho adicional do thread
        }));
    final_tarefas.wait(); // 5: Bloqueia até que todos os threads decrementem o latch
    processar_dados_finais(dados_partes, num_threads_config); // 6: Processa dados após todos os threads concluírem
} // 7: Fim da função

Sincronização com Barreiras (std::barrier)

std::barrier é mais simples e tem menor sobrecarga, enquanto std::flex_barrier é mais flexível, mas com maior sobrecarga.

Exemplo de sincronização de threads com std::barrier:

BlocoResultado processar_bloco(BlocoDados);
std::vector<BlocoDados>
    dividir_em_pedacos(ConjuntoDados dados, unsigned num_threads);

void processar_conjunto_dados(FonteDados &fonte, SinkDados &destino) {
    unsigned const concorrencia_hw = std::thread::hardware_concurrency();
    unsigned const numero_de_threads = (concorrencia_hw > 0) ? concorrencia_hw : 2;
    std::experimental::barrier sincronia_barreira(numero_de_threads);
    std::vector<JoiningThread> threads_trabalho(numero_de_threads); // JoiningThread é um wrapper para std::thread
    std::vector<BlocoDados> pedacos;
    BlocoResultado resultado_bloco_total;

    for (unsigned i = 0; i < numero_de_threads; ++i) {
        threads_trabalho[i] = JoiningThread([&, i] {
            while (!fonte.concluido()) { // 6: Loop principal de processamento
                if (!i) { // 1: Apenas o primeiro thread divide os dados
                    ConjuntoDados bloco_atual = fonte.obter_proximo_bloco_dados();
                    pedacos = dividir_em_pedacos(bloco_atual, numero_de_threads);
                }
                sincronia_barreira.arrive_and_wait(); // 2: Todos os threads esperam aqui
                resultado_bloco_total.set_pedaco(i, numero_de_threads, processar_bloco(pedacos[i])); // 3: Processa o pedaço de dados
                sincronia_barreira.arrive_and_wait(); // 4: Todos os threads esperam aqui novamente
                if (!i) { // 5: Apenas o primeiro thread escreve o resultado
                    destino.escrever_dados(std::move(resultado_bloco_total));
                }
            }
        });
    }
} // 7: Fim da função

std::flex_barrier é mais flexível, com um construtor adicional que aceita uma função completa e o número de threads. Quando todos os threads chegam à barreira, essa função é executada por um deles. Ela também oferece uma maneira de modificar o número de threads que participarão do próximo ciclo da barreira.

Exemplo de sincronização de threads com std::flex_barrier:

void processar_dados_flex(FonteDados &fonte, SinkDados &destino) {
    unsigned const concorrencia_hw = std::thread::hardware_concurrency();
    unsigned const numero_de_threads = (concorrencia_hw > 0) ? concorrencia_hw : 2;
    std::vector<BlocoDados> pedacos_dados;

    auto dividir_fonte = [&] { // 1: Função para dividir a fonte de dados
        if (!fonte.concluido()) {
            ConjuntoDados bloco_atual = fonte.obter_proximo_bloco_dados();
            pedacos_dados = dividir_em_pedacos(bloco_atual, numero_de_threads);
        }
    };
    dividir_fonte(); // 2: Primeira divisão antes do loop

    BlocoResultado resultado_agregado;
    std::experimental::flex_barrier sincronia_flex(numero_de_threads, [&] { // 3: Barreira flexível
        destino.escrever_dados(std::move(resultado_agregado)); // Escreve o resultado
        dividir_fonte(); // 4: Divide a próxima parte da fonte
        return -1; // 5: Retorna -1 para manter o número de threads
    });

    std::vector<JoiningThread> threads_flex_trabalho(numero_de_threads);
    for (unsigned i = 0; i < numero_de_threads; ++i) {
        threads_flex_trabalho[i] = JoiningThread([&, i] {
            while (!fonte.concluido()) { // 6: Loop de trabalho
                resultado_agregado.set_pedaco(i, numero_de_threads, processar_bloco(pedacos_dados[i]));
                sincronia_flex.arrive_and_wait(); // 7: Chega e espera na barreira
            }
        });
    }
}

Um retorno de -1 da função da barreira significa que o número de threads permanece inalterado. Um retorno de 0 ou outro valor especifica o número de threads participantes na próxima iteração.

Operações Atômicas ☢️

Uma operação atômica é uma operação indivisível, a menor unidade de operação. Se uma operação de carregamento (leitura) de um objeto é atômica, todas as operações de modificação desse objeto também devem ser atômicas.

Tipos atômicos não podem ser copiados ou atribuídos via cópia. A cópia e a atribuição envolvem a leitura do valor de um objeto e sua escrita em outro. Para dois objetos independentes, isso são duas operações independentes e, portanto, não atômicas. Consequentemente, essas operações não são permitidas.

Os tipos atômicos padrão são definidos no cabeçalho <atomic>. Quase todas as operações atômicas possuem uma função membro is_lock_free(), que permite ao usuário verificar se a operação atômica é implementada diretamente com instruções atômicas (x.is_lock_free() retorna true) ou se utiliza uma estrutura de lock interna (retorna false).

Operações atômicas podem substituir mutexes para sincronização. Se houver uma necessidade de operações atômicas, é preferível usar uma implementação que não seja baseada em mutexes.

No C++17, todos os tipos atômicos possuem uma variável membro static constexpr chamada is_always_lock_free, que retorna true se o tipo atômico X for lock-free no hardware correspondente.

Apenas o tipo std::atomic_flag não fornece is_lock_free(). Este tipo é uma simples flag booleana, e as operações nele são sempre lock-free.

Tipos Atômicos Padrão e Suas Especializações de std::atomic<>

Para tipos padrão T, o tipo atômico correspondente é atomic_T. Isso se aplica a tipos embutidos, com abreviações para signed (s), unsigned (u) e long long (llong). Para o template std::atomic, é preferível usar a especialização com o tipo T apropriado em vez de usar os aliases.

Tipos atômicos padrão geralmente não podem ser copiados ou atribuídos por cópia; eles não possuem construtores de cópia ou operadores de atribuição de cópia. No entanto, podem ser implicitamente convertidos para seus tipos embutidos correspondentes, suportando operações como load(), store(), exchange(), compare_exchange_weak() e compare_exchange_strong().

O template de classe std::atomic<> não é apenas um conjunto de tipos especializáveis; ele também pode ser usado como um template genérico para criar variáveis atômicas de tipos personalizados. Nesses casos, as operações são limitadas a load(), store() (atribuição e conversão para o tipo do usuário), exchange(), compare_exchange_weak() e compare_exchange_strong().

Operações atômicas são divididas em três categorias:

• Operações de Store, com as seguintes ordens de memória opcionais: memory_order_relaxed, memory_order_release, memory_order_seq_cst.
• Operações de Load, com as seguintes ordens de memória opcionais: memory_order_relaxed, memory_order_consume, memory_order_acquire, memory_order_seq_cst.
• Operações de Read-Modify-Write (RMW), com as seguintes ordens de memória opcionais: memory_order_relaxed, memory_order_consume, memory_order_acquire, memory_order_release, memory_order_acq_rel, memory_order_seq_cst.

Para mais detalhes, consulte o artigo sobre modelos e ordens de memória.

Consistência Sequencial (Sequential Consistency)

Corresponde ao modelo de consistência sequencial, com o símbolo de restrição memory_order_seq_cst, é o modelo de memória mais rigoroso para granularidade de controle e é o modelo padrão para std::atomic.

• A ordem de execução de cada thread é estritamente consistente com a ordem do código.
• A ordem de execução entre threads pode ser intercalada, mas para um único thread, a execução ainda é sequencial.
• Todas as operações de consistência sequencial ocorrem na ordem especificada pelo código, o que se alinha com a lógica do desenvolvedor. No entanto, essa ordenação estrita limita a capacidade dos CPUs modernos de aproveitar o paralelismo de hardware, impactando severamente o desempenho do sistema.

Relaxado (Relaxed)

Corresponde a memory_order_relaxed. Este tipo garante apenas que o acesso aos dados é atômico (não será interrompido por operações de outros threads), mas não impõe nenhuma restrição na ordem de acesso à memória. Leitura e escrita de dados diferentes podem ser reordenadas. É frequentemente usado para contagem estatística.

Acquire-Release

Corresponde a memory_order_consume, memory_order_acquire, memory_order_release e memory_order_acq_rel.

• Acquire: Se uma operação X tiver semântica de acquire, todas as instruções de leitura e escrita após a operação X não serão reordenadas para antes da operação X.
• Release: Se uma operação X tiver semântica de release, todas as instruções de leitura e escrita antes da operação X não serão reordenadas para depois da operação X.

Considerando uma variável atômica A, a sincronização entre uma operação de escrita (Release) em A e uma operação de leitura (Acquire) em A estabelece restrições de ordenação. Para uma escrita (release) X, todas as instruções de leitura e escrita antes de X não podem ser movidas para depois de X. Para uma leitura (acquire) Y, todas as instruções de leitura e escrita após Y não podem ser movidas para antes de Y.

memory_order_release

Se uma operação de escrita X em uma variável atômica A é marcada com memory_order_release, então no thread atual T1, nenhuma operação de leitura/escrita antes de X pode ser reordenada para depois de X. Quando outro thread T2 lê a variável atômica A com memory_order_acquire, todas as operações de leitura/escrita que ocorreram antes de X no thread T1 se tornam visíveis para o thread T2. Se T2 lê com memory_order_consume, apenas as operações de leitura/escrita nas quais A "depende" se tornam visíveis (operações de memória sem dependência não garantem ordem).

É importante notar que para uma escrita com memory_order_release, a garantia de que as operações de leitura/escrita anteriores não serão reordenadas para depois da escrita só vale se outro thread realizar uma leitura dessa variável atômica com memory_order_acquire ou memory_order_consume.

Para uma operação de carga (load) em uma variável atômica com memory_order_acquire: no thread atual, nenhuma operação de leitura/escrita após a carga pode ser reordenada para antes dela. Se outro thread usar memory_order_release para uma operação de store nesta variável atômica, essa operação de store e tudo que a precede se torna visível no thread atual.

Em outras palavras, do ponto de vista do thread T2, todas as gravações na memória que ocorreram antes da operação de gravação atômica em A serão visíveis no thread T1. Ou seja, uma vez que a leitura atômica é concluída, o thread T1 tem a garantia de ver todo o conteúdo que o thread A gravou na memória.

#include <thread>
#include <atomic>
#include <cassert>
#include <string>
 
std::atomic<std::string*> ptr_compartilhado;
int dado_compartilhado;
 
void funcao_produtora() {
  std::string* p  = new std::string("Olá Concorrência");  // L10
  dado_compartilhado = 101; // L11
  ptr_compartilhado.store(p, std::memory_order_release); // L12
}
 
void funcao_consumidora() {
  std::string* p2;
  while (!(p2 = ptr_compartilhado.load(std::memory_order_acquire))); // L17
  assert(*p2 == "Olá Concorrência"); // L18
  assert(dado_compartilhado == 101); // L19
}
 
int main() {
  std::thread t1(funcao_produtora);
  std::thread t2(funcao_consumidora);
  t1.join(); 
  t2.join();
  
  return 0;
}

Neste exemplo, a escrita atômica em ptr_compartilhado (L12) usa memory_order_release. Isso significa que em funcao_produtora, as operações L10 e L11 não serão reordenadas para depois de L12. Em funcao_consumidora, a leitura de ptr_compartilhado (L17) usa memory_order_acquire. Isso implica que L18 e L19 não serão reordenadas para antes de L17. Assim, quando L17 lê um ptr_compartilhado não nulo, as operações L10 e L11 no thread produtor se tornam visíveis para o thread consumidor, garantindo que os assert em funcao_consumidora sejam verdadeiros.

memory_order_consume

Uma operação de carga (load) que usa memory_order_consume: no thread atual, todas as operações de leitura e escrita que dependem desta variável atômica e ocorrem após a carga não podem ser reordenadas para antes dela. Se outro thread usa memory_order_release para uma operação de store nesta variável atômica, essa operação e as leituras/escritas dependentes anteriores se tornam visíveis no thread atual.

Para entender memory_order_consume, precisamos entender as relações de dependência. Exemplo:

std::atomic<std::string*> ponteiro_at;
int valor_nao_at;

std::string* p_str = new std::string("FooBar");
valor_nao_at = 99;                                   
ponteiro_at.store(p_str, std::memory_order_release);

Neste exemplo, ponteiro_at depende de p_str, mas não de valor_nao_at. p_str e valor_nao_at não dependem um do outro.

Com memory_order_consume, se uma escrita em uma variável atômica em T1 é marcada com memory_order_release, e uma leitura dessa mesma variável em T2 é marcada com memory_order_consume, então do ponto de vista de T1, apenas as operações de leitura/escrita que têm uma relação de dependência com essa variável antes da escrita atômica são garantidas como não reordenadas para depois da escrita. Ao contrário de memory_order_acquire, que impõe uma ordem mais forte para todas as operações de leitura/escrita anteriores.

#include <thread>
#include <atomic>
#include <cassert>
#include <string>
 
std::atomic<std::string*> ptr_msg;
int id_msg;
 
void enviar_mensagem() {
  std::string* p_data  = new std::string("Mensagem Importante"); // L10
  id_msg = 505; // L11
  ptr_msg.store(p_data, std::memory_order_release); // L12
}
 
void receber_mensagem() {
  std::string* p_received;
  while (!(p_received = ptr_msg.load(std::memory_order_consume))); // L17
  assert(*p_received == "Mensagem Importante"); // L18
  assert(id_msg == 505); // L19 (Pode falhar!)
}
 
int main() {
  std::thread th_sender(enviar_mensagem);
  std::thread th_receiver(receber_mensagem);
  th_sender.join(); 
  th_receiver.join();
  
  return 0;
}

Comparado ao exemplo de memory_order_acquire, a função enviar_mensagem() permanece inalterada, mas em receber_mensagem(), a operação de carga usa memory_order_consume. Essa mudança pode ter as seguintes implicações: em enviar_mensagem(), ptr_msg tem uma relação de dependência com p_data, então p_data não será reordenado para depois da operação store() (L12). No entanto, id_msg não tem uma relação de dependência com ptr_msg, então pode ser reordenado para depois de L12, o que pode fazer com que o assert() em L19 falhe.

Quando a restrição de ordem de acesso à memória é necessária apenas para variáveis específicas, memory_order_consume deve ser preferido para reduzir as restrições de reordenação e melhorar o desempenho.

memory_order_acq_rel

Equivalente ao uso simultâneo de memory_order_release e memory_order_acquire. Entende-se que as operações antes desta atômica podem ser reordenadas para antes dela, e as operações depois desta atômica podem ser reordenadas para depois dela.

std::atomic_flag: O Tipo Atômico Mais Básico

Objetos deste tipo podem alternar entre dois estados: definido (set) e limpo (clear). Objetos std::atomic_flag devem ser inicializados com ATOMIC_FLAG_INIT. O estado inicial da flag é sempre "limpo".

Uma vez inicializado, um objeto std::atomic_flag pode ser destruído, limpo (clear()), ou definido (test_and_set(), que define e retorna o valor anterior).

Suas características limitadas tornam std::atomic_flag ideal para spinlocks. A flag é inicializada como "limpa", e o mutex está desbloqueado. Para bloquear o mutex, um loop executa test_and_set() até que o valor antigo seja false, o que significa que este thread conseguiu definir a flag para true. Desbloquear o mutex é simplesmente limpar a flag.

std::atomic<T*>

Definido por uma especialização de std::atomic, as operações são sobre ponteiros do tipo T. Embora não possa ser copiado ou atribuído, pode ser construído e atribuído com ponteiros do tipo adequado. std::atomic<T*> também possui as funções membro load(), store(), exchange(), compare_exchange_weak() e compare_exchange_strong(), que recebem e retornam valores do tipo T*.

std::atomic<T*> oferece novas operações para aritmética de ponteiros. As operações básicas são fetch_add() e fetch_sub(), que realizam adição e subtração atômica no endereço armazenado, fornecendo um wrapper simples para +=, -=, ++ e --.

class Entidade{};
Entidade array_entidades[5];
std::atomic<Entidade*> ponteiro_entidade(array_entidades);

Entidade* x = ponteiro_entidade.fetch_add(2); // Adiciona 2 ao ponteiro e retorna o valor original
assert(x == array_entidades);
assert(ponteiro_entidade.load() == &array_entidades[2]);

x = (ponteiro_entidade -= 1); // Subtrai 1 do ponteiro e retorna o valor original
assert(x == &array_entidades[1]);
assert(ponteiro_entidade.load() == &array_entidades[1]);

Operações Relacionadas a Tipos Atômicos Inteiros Padrão

Assim como o conjunto de operações comuns (load(), store(), exchange(), compare_exchange_weak() e compare_exchange_strong()), std::atomic<integral_type> e std::atomic<floating_point_type> possuem um conjunto completo de operações disponíveis: fetch_add(), fetch_sub(), fetch_and(), fetch_or(), fetch_xor(), bem como operadores de atribuição compostos (+=, -=, &=, |= e ^=) e operadores de incremento/decremento (++x, x++, --x e x--).

A biblioteca padrão C++ também fornece funções não-membro para tipos atômicos que envolvem std::shared_ptr<>. Essas operações sobrecarregam as operações de tipos atômicos padrão e aceitam std::shared_ptr<>* como o primeiro argumento.

std::shared_ptr<MeusDadosCompartilhados> p_global;

void processar_dados_globais()
{
    std::shared_ptr<MeusDadosCompartilhados> local_data = std::atomic_load(&p_global);
    // ... processa local_data ...
}

void atualizar_dados_globais()
{
    std::shared_ptr<MeusDadosCompartilhados> novo_dado(new MeusDadosCompartilhados);
    std::atomic_store(&p_global, novo_dado);
}

Para implementações de contêineres concorrentes baseados em locks, consulte a documentação de referência.

Para implementações concorrentes lock-free, consulte a documentação de referência.

A programação lock-free exige atenção especial a:

• Liberação de recursos em estruturas lock-free.
• O problema ABA.
• Otimização da ordem de memória em cenários lock-free.

Design Concorrente

A abordagem mais fundamental é pré-dividir os dados, com múltiplos threads processando suas respectivas partes e o thread principal consolidando os resultados.

Exemplo de quicksort paralelo utilizando std::future:

template<typename T>
std::list<T> quick_sort_paralelo_design(std::list<T> entrada)
{
    if(entrada.empty())
    {
        return entrada;
    }
    std::list<T> resultado;
    resultado.splice(resultado.begin(), entrada, entrada.begin());
    T const& pivo = *resultado.begin();
    auto ponto_divisao = std::partition(entrada.begin(), entrada.end(),
                                     [&](T const& t){return t < pivo;});
    std::list<T> parte_inferior;
    parte_inferior.splice(parte_inferior.end(), entrada, entrada.begin(),
                      ponto_divisao);
    std::future<std::list<T> > futuro_menor(
        // 1
        std::async(&quick_sort_paralelo_design<T>, std::move(parte_inferior)));
    auto nova_parte_superior(
        quick_sort_paralelo_design(std::move(entrada)));
    // 2
    resultado.splice(resultado.end(), nova_parte_superior);
    // 3
    resultado.splice(resultado.begin(), futuro_menor.get());
    // 4
    return resultado;
}

Melhorias no Algoritmo

Para evitar a sobrecarga de criação de threads, utilize std::thread::hardware_concurrency() para obter o número de núcleos de hardware disponíveis e crie novos threads apenas se o número atual de threads for menor. Para evitar que os threads fiquem bloqueados esperando resultados, eles devem ser capazes de continuamente obter mais trabalho.

Quando um thread usa std::partition para dividir dados em uma parte_inferior e uma parte_superior, a parte_inferior pode ser empacotada com um std::promise em um "pedaço de trabalho". O std::future associado a este std::promise é retido para sincronização posterior. Este pedaço é então colocado em um contêiner thread-safe compartilhado (por exemplo, uma fila de tarefas).

Se o número atual de threads em execução for menor que a concorrência de hardware, um novo thread pode ser iniciado para processar um pedaço da fila de tarefas da mesma forma, com o resultado armazenado no std::promise do pedaço para sincronização. O thread original, que ainda tem a parte_superior, procede para processá-la recursivamente.

Após o processamento da parte_superior, o thread verifica o std::future::wait_for(std::chrono::second(0)) para ver se a parte_inferior já foi concluída. Se sim, ele a mescla e retorna. Caso contrário, ele pode pegar qualquer pedaço de trabalho da fila de tarefas compartilhada, garantindo a utilização plena dos recursos do thread.

Design de Código Concorrente

A divisão de tarefas pode ser feita com base no tipo da tarefa ou nos dados da tarefa.

Contenção de Dados e Cache Ping-Pong

Quando dois threads em processadores diferentes leem os mesmos dados, eles geralmente copiam esses dados para seus caches locais e os processam simultaneamente. No entanto, se um thread modifica os dados, essa alteração precisa ser atualizada nos caches de outros núcleos. Isso pode levar tempo e potencialmente parar outros processadores enquanto aguardam a atualização dos dados na memória principal e nos caches. Isso é conhecido como cache ping-pong.

std::atomic<unsigned long> contador(0);

void loop_processamento()
{
    while(contador.fetch_add(1, std::memory_order_relaxed) < 100000000)
    {
        executar_subtarefa();
    }
}

A variável contador é global, acessível por todos os threads que chamam loop_processamento(). Cada incremento requer que o processador garanta que o valor mais recente de contador está em seu cache antes de modificá-lo. Se outro thread estiver executando o mesmo código em outro processador, os dados de contador precisarão ser transferidos entre os caches dos dois processadores, contendo o valor mais recente. Se muitos processadores estiverem executando este código simultaneamente, eles podem se esperar mutuamente. Esta situação é chamada de alta contenção. Se os processadores raramente se esperam, é baixa contenção. No loop, os dados do contador são passados entre os caches muitas vezes, um fenômeno conhecido como cache ping-pong. O mesmo ocorre com mutexes em alta contenção.

Falso Compartilhamento (False Sharing)

Caches de processadores geralmente usam linhas de cache (cache lines) em vez de armazenar dados individualmente. O tamanho de um bloco de memória é determinado pelo processador. Dados menores são armazenados em posições adjacentes na mesma linha de cache. Quando um conjunto de dados acessado por um thread está dentro da mesma linha de cache, o desempenho é superior ao acesso a várias linhas de cache. No entanto, problemas de desempenho surgem quando uma linha de cache contém dados não relacionados que são acessados por threads diferentes.

Considere um array de inteiros onde um grupo de threads acessa elementos do array com muita frequência. Tipicamente, o tamanho de um int é menor que uma linha de cache. Sob o protocolo MESI, mesmo que threads modifiquem elementos diferentes dentro da mesma linha de cache, eles competirão pela propriedade dessa linha, levando ao cache ping-pong em alta contenção. Esta linha de cache é então considerada em estado de falso compartilhamento. Para evitar isso, os dados podem ser estruturados em linhas de cache diferentes. O cabeçalho <new> do C++17 define std::hardware_destructive_interference_size, que especifica o número máximo de bytes contíguos que podem ser compartilhados. Garantir que o espaçamento entre os dados seja maior ou igual a esse valor pode prevenir o falso compartilhamento.

Organizando Dados de Forma Compacta

O falso compartilhamento ocorre porque os dados acessados por um thread estão muito próximos dos dados de outro thread. No entanto, usar dados não compactos pode causar latência de acesso para um único thread. Para evitar o falso compartilhamento, diferentes threads devem acessar diferentes linhas de memória. Dentro de um único thread, os acessos devem ser o mais compactos possível para maximizar a taxa de acerto do cache.

Para testar o falso compartilhamento, pode-se preencher blocos de dados grandes para forçar a distribuição dos dados em diferentes linhas de cache e, em seguida, acessá-los concorrentemente. Se isso melhorar o desempenho, sugere-se que o falso compartilhamento estava ocorrendo.

struct DadosProtegidos
{
    std::mutex m;
    char preenchimento[65536]; // Ou std::hardware_destructive_interference_size para ocupar mais de uma linha de memória
    MeusDados dados_a_proteger;
}; // Para testar falso compartilhamento em contenção de mutex

struct MeusDados
{
    ItemDado item1;
    ItemDado item2;
    char preenchimento[65536];
};
MeusDados algum_array[256]; // Para testar falso compartilhamento em dados de array

Design de Estruturas de Dados para Multithreading

Ao projetar estruturas de dados para desempenho multithread, os fatores cruciais a serem considerados são contenção, falso compartilhamento e proximidade de dados, que têm um impacto significativo no desempenho.

Ao projetar estruturas de dados para multithreading, deve-se tentar:

• Ajustar a distribuição dos dados entre threads para que os dados do mesmo thread estejam intimamente relacionados.
• Reduzir a quantidade de dados necessários em cada thread.
• Fazer com que diferentes threads acessem diferentes locais de armazenamento para evitar o falso compartilhamento.

Segurança contra Exceções em Algoritmos Paralelos

Algoritmos paralelos geralmente exigem mais atenção à segurança contra exceções do que os algoritmos sequenciais. Em um algoritmo sequencial, se uma operação lança uma exceção, o algoritmo só precisa se preocupar em se limpar para evitar vazamentos de recursos e a corrupção de invariantes, e a exceção pode ser propagada para o chamador. Em algoritmos paralelos, muitas operações precisam ser executadas em threads independentes, que não podem propagar exceções diretamente. Se uma função sair com uma exceção após criar um thread, a aplicação pode terminar.

Evite o uso de std::terminate onde for possível.

Se o objetivo é que o novo thread execute uma tarefa e retorne um resultado, e o código pode lançar exceções, std::packaged_task e std::future podem ser combinados (veja o std::accumulate paralelo usando std::packaged_task).

try
{
    for(unsigned long i=0; i<(num_threads-1); ++i)
    {
        // ... como antes
    }
    T ultimo_resultado = acumular_bloco_func()(inicio_bloco, ultimo);
    std::for_each(threads_execucao.begin(), threads_execucao.end(),
                  std::mem_fn(&std::thread::join));
}
catch(...)
{
    for(unsigned long i=0; i<(num_threads-1); ++i)
    {
        if(threads_execucao[i].joinable())
            threads_execucao[i].join();
    }
    throw;
}

Use std::future para capturar exceções; elas serão relançadas quando std::future::get() for chamado. A exceção pode ser tratada na camada externa, garantindo que os threads sejam aguardados para evitar vazamentos de threads (threads não joined()). Classes RAII também podem ser usadas para controlar threads, garantindo que todos os threads joinable sejam aguardados quando o objeto RAII é destruído.

Além disso, se um objeto std::future for destruído sem ser aguardado, seu destrutor esperará que o thread correspondente termine. Isso pode mascarar problemas de vazamento de threads, pois o thread ainda está em execução e mantém referências a dados. A seguir, uma implementação de segurança contra exceções usando std::async().

std::async() também pode ser usado para implementar segurança contra exceções:

template<typename Iterador, typename T>
T acumular_paralelo(Iterador primeiro, Iterador ultimo, T valor_inicial)
{
    unsigned long const tamanho = std::distance(primeiro, ultimo);
    // 1
    unsigned long const tamanho_max_bloco = 25;
    if(tamanho <= tamanho_max_bloco)
    {
        return std::accumulate(primeiro, ultimo, valor_inicial);
        // 2
    }
    else
    {
        Iterador ponto_medio = primeiro;
        std::advance(ponto_medio, tamanho / 2);
        // 3
        std::future<T> resultado_primeira_metade =
            std::async(acumular_paralelo<Iterador,T>,
                       // 4
                       primeiro, ponto_medio, valor_inicial);
        T resultado_segunda_metade = acumular_paralelo(ponto_medio, ultimo, T());
        return resultado_primeira_metade.get() + resultado_segunda_metade;
        // 6
    }
}

Esta versão utiliza std::async para gerar futures, que capturam exceções. Quando get() é chamado, as exceções armazenadas no future são relançadas, permitindo o tratamento na camada externa.

Escalabilidade e a Lei de Amdahl

A escalabilidade é igualmente importante na programação paralela, representando a capacidade de uma aplicação de utilizar plenamente os processadores de um sistema para executar tarefas. Uma aplicação estritamente single-thread é completamente não escalável; adicionar 100 processadores não melhorará o desempenho.

Uma forma de calcular o desempenho de programas multithread é através do seguinte modelo simplificado: dividir o programa em partes "sequenciais" e "paralelas". A parte sequencial é o trabalho que só pode ser executado por um único thread. A parte paralela é a parte onde todos os processadores disponíveis podem trabalhar juntos. Sob essa hipótese, a proporção do tempo gasto na parte "sequencial" do programa é denotada por fs, e o ganho de desempenho (P) com N processadores é estimado por:

$$ P = \frac{1}{fs + \frac{1-fs}{N}} $$

Esta é a Lei de Amdahl. Ela estabelece que maximizar o paralelismo no código garante que todos os processadores possam ser usados para trabalho útil. Reduzir a parte "sequencial" ou o tempo de espera dos threads pode render mais ganhos de desempenho em sistemas multiprocessadores.

Escalabilidade significa reduzir o tempo de execução de uma única ação ou realizar mais trabalho em um determinado tempo quando mais processadores são adicionados. Às vezes, esses dois indicadores são equivalentes.

Latência Oculta em Multithreading

Threads nem sempre estão realizando trabalho útil; às vezes, eles esperam por outros threads, pela conclusão de operações de I/O ou por outros eventos. Se um thread está esperando e há tarefas essenciais a serem concluídas no sistema, essa espera pode ser "ocultada".

Independentemente do motivo da espera, um thread bloqueado significa que ele está esperando por um tempo de CPU. O processador executará outro thread durante o tempo de bloqueio, em vez de ficar ocioso. Portanto, quando se sabe que alguns threads precisarão esperar por um tempo considerável, pode-se aproveitar o tempo ocioso da CPU para executar um ou mais outros threads.

Projetando Código Concorrente

std::for_each Paralelo

O princípio de std::for_each é simples: aplicar uma função fornecida pelo usuário a cada elemento de um determinado intervalo. A principal diferença entre as versões paralela e sequencial é a ordem de chamada da função.

Para implementar uma versão paralela, é necessário dividir os elementos a serem processados por cada thread. Se o número de elementos for conhecido, os dados podem ser divididos antes do processamento. Se apenas tarefas paralelas estiverem em execução, std::thread::hardware_concurrency() pode ser usado para determinar o número de threads. Da mesma forma, se esses elementos puderem ser processados de forma independente, blocos de dados contíguos podem ser usados para evitar falso compartilhamento. Para propagar exceções para o chamador, std::packaged_task e std::future são necessários para transferir exceções entre threads.

std::find Paralelo

(Consulte o arquivo de código para implementações.)

Gerenciamento Avançado de Threads

Pool de Threads

Uma implementação simples de um pool de threads possui um número fixo de threads de trabalho (geralmente igual a std::thread::hardware_concurrency()). Quando há trabalho a ser feito, as tarefas podem ser colocadas em uma fila de tarefas. Cada thread de trabalho pega uma tarefa da fila, a executa e retorna para pegar uma nova tarefa. Os threads no pool não precisam esperar uns pelos outros para concluir tarefas correspondentes. Se a espera for necessária, a sincronização deve ser gerenciada.

Implementação de código de um pool de threads simples:

class PoolDeThreads
{
    std::atomic_bool finalizado;
    ThreadSafeQueue<std::function<void()> > fila_trabalho;
    // 1
    std::vector<std::thread> threads_trabalhadoras;
    // 2
    JoiningThreadWrapper gerenciador_joins; // Classe RAII para fazer join em todos os threads
    // 3
    void thread_trabalhadora()
    {
        while(!finalizado) // 4
        {
            std::function<void()> tarefa;
            if(fila_trabalho.try_pop(tarefa)) // 5: Tenta pegar uma tarefa
            {
                tarefa(); // 6: Executa a tarefa
            }
            else
            {
                std::this_thread::yield(); // 7: Reduz a disputa por CPU
            }
        }
    }
public:
    PoolDeThreads():
        finalizado(false), gerenciador_joins(threads_trabalhadoras)
    {
        unsigned const num_threads = std::thread::hardware_concurrency(); // 8
        try
        {
            for(unsigned i = 0; i < num_threads; ++i)
            {
                threads_trabalhadoras.push_back(
                    std::thread(&PoolDeThreads::thread_trabalhadora, this)); // 9
            }
        }
        catch(...)
        {
            finalizado = true; // 10: Garante sinalização de término em caso de erro
            throw;
        }
    }
    ~PoolDeThreads()
    {
        finalizado = true; // 11: Sinaliza término para todos os threads
    }
    template<typename FunctionType>
    void enviar(FunctionType f)
    {
        fila_trabalho.push(std::function<void()>(f)); // 12: Envia uma tarefa
    }
};

Este pool de threads utiliza uma fila thread-safe para gerenciar tarefas. Os usuários encapsulam as tarefas em std::function<void()> e as empurram para a fila. Os threads de trabalho (thread_trabalhadora) obtêm e executam essas tarefas em um loop, até que a flag finalizado seja definida. Se não houver tarefas na fila, std::this_thread::yield() é chamado para que o thread libere o tempo de CPU, permitindo que outros threads adicionem tarefas à fila.

No entanto, um pool de threads tão simples não consegue lidar com tarefas que retornam valores ou que exigem operações de bloqueio, e pode levar a deadlocks. Em casos simples, std::async geralmente oferece melhor funcionalidade.

Aguardando Tarefas em um Pool de Threads

Para retornar o resultado de uma tarefa de um thread de execução para um thread de espera, é necessário usar futures. Contudo, instâncias de std::packaged_task<> não são copiáveis, apenas móveis. Portanto, std::function<> não pode ser usado para a fila de tarefas, pois std::function<> exige que os objetos de função sejam copiáveis.

Nesse cenário, precisamos de uma classe de "type erasure" personalizada com um operador de chamada de função, que lide com funções sem parâmetros e sem retorno usando chamadas de funções virtuais simples.

Implementação da classe FunctionWrapper:

class FunctionWrapper
{
    struct ImplementacaoBase {
        virtual void chamar() = 0;
        virtual ~ImplementacaoBase() {}
    };
    
    std::unique_ptr<ImplementacaoBase> implementacao;

    template<typename F>
    struct ImplementacaoConcreta: ImplementacaoBase
    {
        F funcao_interna;
        ImplementacaoConcreta(F&& f_): funcao_interna(std::move(f_)) {}
        void chamar() override { funcao_interna(); }
    };
    
public:
    template<typename F>
    FunctionWrapper(F&& f):
        implementacao(new ImplementacaoConcreta<F>(std::move(f)))
    {}
	
    void operator()(){ implementacao->chamar(); }
    
    FunctionWrapper() = default;
    
    FunctionWrapper(FunctionWrapper&& outra) noexcept
    {
        implementacao = std::move(outra.implementacao);
    }
    
    FunctionWrapper(const FunctionWrapper&)=delete; // Impede cópia
    FunctionWrapper& operator=(const FunctionWrapper&)=delete; // Impede atribuição
    FunctionWrapper& operator=(FunctionWrapper&& outra) noexcept
    {
        implementacao = std::move(outra.implementacao);
        return *this;
    }
};

Roubo de Tarefas (Work Stealing)

Quando a fila de trabalho de um thread está vazia, ele entra em estado de espera, desperdiçando recursos. É ideal manter cada thread de trabalho o mais saturado possível.

Cada fila de trabalho de thread local é armazenada em uma lista global de variáveis do pool de threads.

class PoolDeThreadsComRoubo
{
    // ... membros existentes ...
    std::vector<std::unique_ptr<FilaDeRouboDeTrabalho>> filas_locais;
    // Cada fila de trabalho de thread é armazenada em uma lista global no pool.
    
    // Usado para indexar as filas
    static thread_local FilaDeRouboDeTrabalho* fila_trabalho_local;
    static thread_local unsigned meu_indice_local;
    
    void thread_trabalhadora(unsigned indice_do_thread)
    {
        meu_indice_local = indice_do_thread;
        // O novo thread de trabalho inicializa a variável local_thread com o índice dado pelo pool
        fila_trabalho_local = filas_locais[meu_indice_local].get();
        // Obtém a fila de trabalho com base no índice
        while(!finalizado)
        {
            executar_tarefa_pendente();
        }
    }
public:
    PoolDeThreadsComRoubo()
    {
        unsigned const num_threads = std::thread::hardware_concurrency();
        
        for(unsigned i = 0; i < num_threads; ++i)
        {
            filas_locais.push_back(std::unique_ptr<FilaDeRouboDeTrabalho>(new FilaDeRouboDeTrabalho));
            // Atribui um índice ao criar cada thread
            threads_trabalhadoras.push_back(std::thread(&PoolDeThreadsComRoubo::thread_trabalhadora, this, i));
        }
    }
    
    void executar_tarefa_pendente()
    {
        FunctionWrapper tarefa;
        
        if(pegar_tarefa_da_fila_local(tarefa) ||        // Pega tarefa da fila local
           pegar_tarefa_da_fila_global(tarefa) ||       // Pega tarefa da fila global do pool
           roubar_tarefa_de_outra_fila(tarefa))       // Rouba tarefa da fila local de outro thread
        {
            tarefa();
        }
        else
        {
            std::this_thread::yield();
        }
    }
    
    // Interface de submissão de tarefas
    template<typename FunctionType>
    std::future<typename std::result_of<FunctionType()>::type>
        enviar_tarefa(FunctionType f)
    {
        typedef typename std::result_of<FunctionType()>::type tipo_resultado;
        std::packaged_task<tipo_resultado()> tarefa(f);
        std::future<tipo_resultado> res(tarefa.get_future());
        // Verifica se é um thread de trabalho
        if(fila_trabalho_local) // Se for um thread do pool, adiciona na sua fila local
        {
            fila_trabalho_local->push(std::move(tarefa));
        }
        else // Se não for um thread do pool, adiciona na fila global
        {
            fila_trabalho_global_pool.push(std::move(tarefa));
        }
        return res;
    }
    
    // ... outros métodos do pool ...
};