Pool de threads em C ++ 11

Questões relevantes :

Sobre o C ++ 11:

• C ++ 11: std :: thread em pool?
• O async (launch :: async) no C ++ 11 tornará obsoletos os pools de threads para evitar a criação cara de threads?

Sobre o Boost:

• C ++ boost thread reutilizando threads
• boost :: thread e criando um pool deles!

Como obtenho um pool de threads para enviar tarefas para , sem criar e excluí-las repetidamente? Isso significa threads persistentes para ressincronizar sem ingressar.

Eu tenho um código que se parece com isso:

namespace {
  std::vector<std::thread> workers;

  int total = 4;
  int arr[4] = {0};

  void each_thread_does(int i) {
    arr[i] += 2;
  }
}

int main(int argc, char *argv[]) {
  for (int i = 0; i < 8; ++i) { // for 8 iterations,
    for (int j = 0; j < 4; ++j) {
      workers.push_back(std::thread(each_thread_does, j));
    }
    for (std::thread &t: workers) {
      if (t.joinable()) {
        t.join();
      }
    }
    arr[4] = std::min_element(arr, arr+4);
  }
  return 0;
}

Em vez de criar e unir threads a cada iteração, prefiro enviar tarefas aos threads de trabalho a cada iteração e criá-las apenas uma vez.

Você pode usar a biblioteca de conjuntos de threads C ++, https://github.com/vit-vit/ctpl .

Em seguida, o código que você escreveu pode ser substituído pelo seguinte

#include <ctpl.h>  // or <ctpl_stl.h> if ou do not have Boost library

int main (int argc, char *argv[]) {
    ctpl::thread_pool p(2 /* two threads in the pool */);
    int arr[4] = {0};
    std::vector<std::future<void>> results(4);
    for (int i = 0; i < 8; ++i) { // for 8 iterations,
        for (int j = 0; j < 4; ++j) {
            results[j] = p.push([&arr, j](int){ arr[j] +=2; });
        }
        for (int j = 0; j < 4; ++j) {
            results[j].get();
        }
        arr[4] = std::min_element(arr, arr + 4);
    }
}

Você obterá o número desejado de threads e não os criará e excluirá repetidamente nas iterações.

Isso é copiado da minha resposta para outro post muito semelhante, espero que possa ajudar:

1) Comece com o número máximo de threads que um sistema pode suportar:

int Num_Threads =  thread::hardware_concurrency();

2) Para uma implementação eficiente do conjunto de encadeamentos, uma vez que os encadeamentos são criados de acordo com o Num_Threads, é melhor não criar novos ou destruir os antigos (ingressando). Haverá uma penalidade de desempenho, pode até tornar seu aplicativo mais lento que a versão serial.

Cada encadeamento C ++ 11 deve estar executando em sua função com um loop infinito, aguardando constantemente novas tarefas serem executadas.

Aqui está como anexar essa função ao pool de threads:

int Num_Threads = thread::hardware_concurrency();
vector<thread> Pool;
for(int ii = 0; ii < Num_Threads; ii++)
{  Pool.push_back(thread(Infinite_loop_function));}

3) A função Infinite_loop

Este é um loop "while (true)" aguardando a fila de tarefas

void The_Pool:: Infinite_loop_function()
{
    while(true)
    {
        {
            unique_lock<mutex> lock(Queue_Mutex);

            condition.wait(lock, []{return !Queue.empty() || terminate_pool});
            Job = Queue.front();
            Queue.pop();
        }
        Job(); // function<void()> type
    }
};

4) Crie uma função para adicionar trabalho à sua fila

void The_Pool:: Add_Job(function<void()> New_Job)
{
    {
        unique_lock<mutex> lock(Queue_Mutex);
        Queue.push(New_Job);
    }
    condition.notify_one();
}

5) Vincule uma função arbitrária à sua fila

Pool_Obj.Add_Job(std::bind(&Some_Class::Some_Method, &Some_object));

Depois de integrar esses ingredientes, você terá seu próprio pool de segmentação dinâmico. Esses encadeamentos sempre são executados, aguardando a execução do trabalho.

Peço desculpas se houver algum erro de sintaxe, digitei esse código e tenho uma memória ruim. Lamento não poder fornecer o código completo do conjunto de encadeamentos, o que violaria a integridade do meu trabalho.

Edit: para finalizar o pool, chame o método shutdown ():

XXXX::shutdown(){
{
    unique_lock<mutex> lock(threadpool_mutex);
    terminate_pool = true;} // use this flag in condition.wait

    condition.notify_all(); // wake up all threads.

    // Join all threads.
    for(std::thread &every_thread : thread_vector)
    {   every_thread.join();}

    thread_vector.clear();  
    stopped = true; // use this flag in destructor, if not set, call shutdown() 
}

Um conjunto de encadeamentos significa que todos os encadeamentos estão em execução o tempo todo - em outras palavras, a função de encadeamento nunca retorna. Para dar aos threads algo significativo a ser feito, é necessário projetar um sistema de comunicação entre threads, com o objetivo de informar ao thread que há algo a ser feito, bem como para comunicar os dados de trabalho reais.

Normalmente, isso envolverá algum tipo de estrutura de dados simultânea, e cada encadeamento presumivelmente dormirá em algum tipo de variável de condição, que será notificada quando houver trabalho a ser feito. Ao receber a notificação, um ou vários threads são ativados, recuperam uma tarefa da estrutura de dados simultânea, processam-na e armazenam o resultado de maneira análoga.

O tópico continuaria verificando se ainda há mais trabalho a ser feito e, se não voltar a dormir.

O resultado é que você mesmo deve projetar tudo isso, pois não existe uma noção natural de "trabalho" que seja universalmente aplicável. É um pouco de trabalho, e há alguns problemas sutis que você precisa corrigir. (Você pode programar no Go se quiser um sistema que cuide do gerenciamento de threads nos bastidores.)

Um conjunto de encadeamentos é, no núcleo, um conjunto de encadeamentos, todos vinculados a uma função que funciona como um loop de eventos. Esses encadeamentos aguardam incessantemente a execução de uma tarefa ou seu próprio término.

O trabalho de conjunto de encadeamentos é fornecer uma interface para enviar trabalhos, definir (e talvez modificar) a política de execução desses trabalhos (regras de planejamento, instanciação de encadeamento, tamanho do conjunto) e monitorar o status dos encadeamentos e recursos relacionados.

Portanto, para um pool versátil, é preciso começar definindo o que é uma tarefa, como é iniciada, interrompida, qual é o resultado (veja a noção de promessa e futuro para essa pergunta), que tipo de eventos os threads terão que responder como eles irão lidar com eles, como esses eventos serão discriminados daqueles tratados pelas tarefas. Isso pode se tornar bastante complicado, como você pode ver, e impor restrições sobre como os encadeamentos funcionarão, pois a solução se torna cada vez mais envolvida.

As ferramentas atuais para lidar com eventos são bastante barebones (*): primitivas como mutexes, variáveis ​​de condição e algumas abstrações além disso (bloqueios, barreiras). Mas, em alguns casos, essas abstrações podem se tornar impróprias (veja esta questão relacionada ), e é preciso voltar a usar as primitivas.

Outros problemas também precisam ser gerenciados:

• sinal
• i / o
• hardware (afinidade do processador, configuração heterogênea)

Como isso aconteceria em seu ambiente?

Esta resposta a uma pergunta semelhante aponta para uma implementação existente destinada ao boost e ao stl.

Ofereci uma implementação muito grosseira de um conjunto de threads para outra pergunta, que não trata de muitos problemas descritos acima. Você pode desenvolver isso. Você também pode querer dar uma olhada nas estruturas existentes em outros idiomas, para encontrar inspiração.

_{(*) Não vejo isso como um problema, muito pelo contrário. Eu acho que é o próprio espírito de C ++ herdado de C.}

Follwoing [PhD EcE](https://stackoverflow.com/users/3818417/phd-ece) suggestion, I implemented the thread pool:

function_pool.h

#pragma once
#include <queue>
#include <functional>
#include <mutex>
#include <condition_variable>
#include <atomic>
#include <cassert>

class Function_pool
{

private:
    std::queue<std::function<void()>> m_function_queue;
    std::mutex m_lock;
    std::condition_variable m_data_condition;
    std::atomic<bool> m_accept_functions;

public:

    Function_pool();
    ~Function_pool();
    void push(std::function<void()> func);
    void done();
    void infinite_loop_func();
};

function_pool.cpp

#include "function_pool.h"

Function_pool::Function_pool() : m_function_queue(), m_lock(), m_data_condition(), m_accept_functions(true)
{
}

Function_pool::~Function_pool()
{
}

void Function_pool::push(std::function<void()> func)
{
    std::unique_lock<std::mutex> lock(m_lock);
    m_function_queue.push(func);
    // when we send the notification immediately, the consumer will try to get the lock , so unlock asap
    lock.unlock();
    m_data_condition.notify_one();
}

void Function_pool::done()
{
    std::unique_lock<std::mutex> lock(m_lock);
    m_accept_functions = false;
    lock.unlock();
    // when we send the notification immediately, the consumer will try to get the lock , so unlock asap
    m_data_condition.notify_all();
    //notify all waiting threads.
}

void Function_pool::infinite_loop_func()
{
    std::function<void()> func;
    while (true)
    {
        {
            std::unique_lock<std::mutex> lock(m_lock);
            m_data_condition.wait(lock, [this]() {return !m_function_queue.empty() || !m_accept_functions; });
            if (!m_accept_functions && m_function_queue.empty())
            {
                //lock will be release automatically.
                //finish the thread loop and let it join in the main thread.
                return;
            }
            func = m_function_queue.front();
            m_function_queue.pop();
            //release the lock
        }
        func();
    }
}

main.cpp

#include "function_pool.h"
#include <string>
#include <iostream>
#include <mutex>
#include <functional>
#include <thread>
#include <vector>

Function_pool func_pool;

class quit_worker_exception : public std::exception {};

void example_function()
{
    std::cout << "bla" << std::endl;
}

int main()
{
    std::cout << "stating operation" << std::endl;
    int num_threads = std::thread::hardware_concurrency();
    std::cout << "number of threads = " << num_threads << std::endl;
    std::vector<std::thread> thread_pool;
    for (int i = 0; i < num_threads; i++)
    {
        thread_pool.push_back(std::thread(&Function_pool::infinite_loop_func, &func_pool));
    }

    //here we should send our functions
    for (int i = 0; i < 50; i++)
    {
        func_pool.push(example_function);
    }
    func_pool.done();
    for (unsigned int i = 0; i < thread_pool.size(); i++)
    {
        thread_pool.at(i).join();
    }
}

Algo assim pode ajudar (extraído de um aplicativo que funciona).

#include <memory>
#include <boost/asio.hpp>
#include <boost/thread.hpp>

struct thread_pool {
  typedef std::unique_ptr<boost::asio::io_service::work> asio_worker;

  thread_pool(int threads) :service(), service_worker(new asio_worker::element_type(service)) {
    for (int i = 0; i < threads; ++i) {
      auto worker = [this] { return service.run(); };
      grp.add_thread(new boost::thread(worker));
    }
  }

  template<class F>
  void enqueue(F f) {
    service.post(f);
  }

  ~thread_pool() {
    service_worker.reset();
    grp.join_all();
    service.stop();
  }

private:
  boost::asio::io_service service;
  asio_worker service_worker;
  boost::thread_group grp;
};

Você pode usá-lo assim:

thread_pool pool(2);

pool.enqueue([] {
  std::cout << "Hello from Task 1\n";
});

pool.enqueue([] {
  std::cout << "Hello from Task 2\n";
});

Lembre-se de que reinventar um mecanismo eficiente de enfileiramento assíncrono não é trivial.

O Boost :: asio :: io_service é uma implementação muito eficiente ou, na verdade, é uma coleção de wrappers específicos da plataforma (por exemplo, envolve as portas de conclusão de E / S no Windows).

Edit: Isso agora requer C ++ 17 e conceitos. (Em 9/12/16, apenas g ++ 6.0+ é suficiente.)

A dedução de modelo é muito mais precisa por causa disso, portanto, vale a pena o esforço de obter um compilador mais novo. Ainda não encontrei uma função que exija argumentos explícitos de modelo.

Agora, também é necessário qualquer objeto que possa ser chamado de maneira apropriada ( e ainda é estaticamente seguro! ).

Agora também inclui um pool de encadeamentos de prioridade de segmentação verde opcional usando a mesma API. Esta classe é apenas POSIX, no entanto. Ele usa a ucontext_tAPI para alternar tarefas do espaço do usuário.

Eu criei uma biblioteca simples para isso. Um exemplo de uso é dado abaixo. (Estou respondendo isso porque foi uma das coisas que encontrei antes de decidir que era necessário escrevê-lo.)

bool is_prime(int n){
  // Determine if n is prime.
}

int main(){
  thread_pool pool(8); // 8 threads

  list<future<bool>> results;
  for(int n = 2;n < 10000;n++){
    // Submit a job to the pool.
    results.emplace_back(pool.async(is_prime, n));
  }

  int n = 2;
  for(auto i = results.begin();i != results.end();i++, n++){
    // i is an iterator pointing to a future representing the result of is_prime(n)
    cout << n << " ";
    bool prime = i->get(); // Wait for the task is_prime(n) to finish and get the result.
    if(prime)
      cout << "is prime";
    else
      cout << "is not prime";
    cout << endl;
  }  
}

Você pode passar asyncqualquer função com qualquer valor de retorno (ou nulo) e qualquer (ou nenhum) argumento e ele retornará um correspondente std::future. Para obter o resultado (ou apenas espere até que uma tarefa seja concluída), você chamaget() o futuro.

Aqui está o github: https://github.com/Tyler-Hardin/thread_pool .

Esta é outra implementação de conjunto de encadeamentos muito simples, fácil de entender e usar, que usa apenas a biblioteca padrão C ++ 11 e pode ser vista ou modificada para seus usos. Deve ser um bom começo para quem deseja usar o encadeamento piscinas:

https://github.com/progschj/ThreadPool

Você pode usar o thread_pool da biblioteca boost:

void my_task(){...}

int main(){
    int threadNumbers = thread::hardware_concurrency();
    boost::asio::thread_pool pool(threadNumbers);

    // Submit a function to the pool.
    boost::asio::post(pool, my_task);

    // Submit a lambda object to the pool.
    boost::asio::post(pool, []() {
      ...
    });
}

Você também pode usar o pool de threads da comunidade de código aberto:

void first_task() {...}    
void second_task() {...}

int main(){
    int threadNumbers = thread::hardware_concurrency();
    pool tp(threadNumbers);

    // Add some tasks to the pool.
    tp.schedule(&first_task);
    tp.schedule(&second_task);
}

Um pool de threads sem dependências fora do STL é totalmente possível. Recentemente, escrevi uma pequena biblioteca de encadeamentos de cabeçalho somente para resolver exatamente o mesmo problema. Ele suporta redimensionamento de pool dinâmico (alterando o número de trabalhadores em tempo de execução), aguardando, parando, pausando, retomando e assim por diante. Espero que você ache útil.