Regroupement de threads en C ++ 11

Questions pertinentes :

À propos de C ++ 11:

• C ++ 11: std :: thread groupé?
• Async (launch :: async) en C ++ 11 rendra-t-il les pools de threads obsolètes pour éviter la création de threads coûteux?

À propos de Boost:

• Thread C ++ boost réutilisant les threads
• boost :: thread et en créer un pool!

Comment puis-je obtenir un pool de threads vers lesquels envoyer des tâches , sans les créer et les supprimer encore et encore? Cela signifie que les threads persistants doivent se resynchroniser sans se joindre.

J'ai du code qui ressemble à ceci:

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

Au lieu de créer et de joindre des threads à chaque itération, je préfère envoyer des tâches à mes threads de travail à chaque itération et ne les créer qu'une seule fois.

Vous pouvez utiliser la bibliothèque de pools de threads C ++, https://github.com/vit-vit/ctpl .

Ensuite, le code que vous avez écrit peut être remplacé par le suivant

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

Vous obtiendrez le nombre de threads souhaité et ne les créerez pas et ne les supprimerez pas encore et encore au fil des itérations.

Ceci est copié de ma réponse dans un autre article très similaire, j'espère que cela pourra vous aider:

1) Commencez avec le nombre maximum de threads qu'un système peut prendre en charge:

int Num_Threads =  thread::hardware_concurrency();

2) Pour une implémentation efficace de threadpool, une fois que les threads sont créés selon Num_Threads, il vaut mieux ne pas en créer de nouveaux, ni détruire les anciens (en les rejoignant). Il y aura une pénalité de performance, peut même ralentir votre application par rapport à la version série.

Chaque thread C ++ 11 doit être exécuté dans sa fonction avec une boucle infinie, attendant constamment de nouvelles tâches à saisir et à exécuter.

Voici comment attacher une telle fonction au 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) La fonction_loop_infini

Il s'agit d'une boucle "while (true)" attendant la file d'attente des tâches

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) Créez une fonction pour ajouter un travail à votre file d'attente

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

5) Liez une fonction arbitraire à votre file d'attente

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

Une fois que vous avez intégré ces ingrédients, vous disposez de votre propre pool de threads dynamique. Ces threads s'exécutent toujours, en attente de travail.

Je m'excuse s'il y a des erreurs de syntaxe, j'ai tapé ces codes et j'ai une mauvaise mémoire. Désolé de ne pas pouvoir vous fournir le code complet du pool de threads, car cela violerait l'intégrité de mon travail.

Edit: pour terminer le pool, appelez la méthode 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() 
}

Un pool de threads signifie que tous vos threads sont en cours d'exécution, tout le temps - en d'autres termes, la fonction thread ne retourne jamais. Pour donner aux threads quelque chose de significatif à faire, vous devez concevoir un système de communication inter-thread, à la fois dans le but de dire au thread qu'il y a quelque chose à faire, ainsi que pour communiquer les données de travail réelles.

En général, cela impliquera une sorte de structure de données concurrente, et chaque thread dormirait probablement sur une sorte de variable de condition, qui serait notifiée lorsqu'il y a du travail à faire. Lors de la réception de la notification, un ou plusieurs des threads se réveillent, récupèrent une tâche à partir de la structure de données simultanée, la traitent et stockent le résultat d'une manière analogue.

Le fil continuerait ensuite de vérifier s'il y avait encore plus de travail à faire, et sinon de se rendormir.

Le résultat est que vous devez concevoir tout cela vous-même, car il n'y a pas de notion naturelle de «travail» qui soit universellement applicable. C'est un peu de travail, et il y a des problèmes subtils que vous devez résoudre. (Vous pouvez programmer dans Go si vous aimez un système qui s'occupe de la gestion des threads pour vous dans les coulisses.)

Un threadpool est au cœur un ensemble de threads tous liés à une fonction fonctionnant comme une boucle d'événements. Ces threads attendront indéfiniment une tâche à exécuter, ou leur propre arrêt.

Le travail de threadpool consiste à fournir une interface pour soumettre des travaux, définir (et peut-être modifier) ​​la politique d'exécution de ces travaux (règles de planification, instanciation des threads, taille du pool) et surveiller l'état des threads et des ressources associées.

Donc pour un pool polyvalent, il faut commencer par définir ce qu'est une tâche, comment elle est lancée, interrompue, quel en est le résultat (voir la notion de promesse et d'avenir pour cette question), à quel genre d'événements les threads devront répondre à, comment ils les traiteront, comment ces événements seront distingués de ceux traités par les tâches. Cela peut devenir assez compliqué comme vous pouvez le voir, et imposer des restrictions sur la façon dont les threads fonctionneront, car la solution devient de plus en plus impliquée.

L'outillage actuel pour gérer les événements est assez simple (*): des primitives comme des mutex, des variables de condition et quelques abstractions en plus (verrous, barrières). Mais dans certains cas, ces abstrations peuvent s'avérer inappropriées (voir cette question connexe ), et il faut revenir à l'utilisation des primitives.

D'autres problèmes doivent également être gérés:

• signal
• i / o
• matériel (affinité du processeur, configuration hétérogène)

Comment cela se jouerait-il dans votre environnement?

Cette réponse à une question similaire pointe vers une implémentation existante destinée au boost et au stl.

J'ai proposé une implémentation très grossière d'un threadpool pour une autre question, qui ne résout pas de nombreux problèmes décrits ci-dessus. Vous voudrez peut-être vous en inspirer. Vous voudrez peut-être également jeter un œil aux frameworks existants dans d'autres langues, pour trouver l'inspiration.

_{(*) Je ne vois pas cela comme un problème, bien au contraire. Je pense que c'est l'esprit même du C ++ hérité 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();
    }
}

Quelque chose comme celui-ci pourrait aider (tiré d'une application qui fonctionne).

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

Vous pouvez l'utiliser comme ceci:

thread_pool pool(2);

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

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

Gardez à l'esprit que réinventer un mécanisme de file d'attente asynchrone efficace n'est pas anodin.

Boost :: asio :: io_service est une implémentation très efficace, ou est en fait une collection de wrappers spécifiques à la plate-forme (par exemple, il enveloppe les ports d'achèvement d'E / S sous Windows).

Edit: Cela nécessite maintenant C ++ 17 et des concepts. (Depuis le 12/09/16, seul g ++ 6.0+ est suffisant.)

La déduction du modèle est beaucoup plus précise à cause de cela, cependant, cela vaut la peine d'obtenir un compilateur plus récent. Je n'ai pas encore trouvé de fonction nécessitant des arguments de modèle explicites.

Il prend également maintenant n'importe quel objet appelable approprié ( et est toujours de type statique !!! ).

Il comprend également désormais un pool de threads de priorité de thread vert facultatif utilisant la même API. Cependant, cette classe est uniquement POSIX. Il utilise l' ucontext_tAPI pour la commutation de tâches en espace utilisateur.

J'ai créé une bibliothèque simple pour cela. Un exemple d'utilisation est donné ci-dessous. (Je réponds à cela parce que c'était l'une des choses que j'ai trouvées avant de décider qu'il était nécessaire de l'écrire moi-même.)

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

Vous pouvez passer asyncn'importe quelle fonction avec n'importe quelle valeur de retour (ou void) et n'importe quel (ou aucun) argument et elle renverra un std::future. Pour obtenir le résultat (ou simplement attendre qu'une tâche soit terminée), vous appelez get()le futur.

Voici le github: https://github.com/Tyler-Hardin/thread_pool .

Ceci est une autre implémentation de pool de threads qui est très simple, facile à comprendre et à utiliser, utilise uniquement la bibliothèque standard C ++ 11, et peut être consultée ou modifiée pour vos utilisations, devrait être un bon début si vous souhaitez utiliser thread piscines:

https://github.com/progschj/ThreadPool

Vous pouvez utiliser thread_pool de la bibliothèque 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, []() {
      ...
    });
}

Vous pouvez également utiliser threadpool de la communauté open source:

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

Un pool de threads sans dépendances en dehors de STL est tout à fait possible. J'ai récemment écrit une petite bibliothèque de threadpool uniquement en-tête pour résoudre exactement le même problème. Il prend en charge le redimensionnement dynamique du pool (modification du nombre de nœuds de calcul au moment de l'exécution), l'attente, l'arrêt, la mise en pause, la reprise, etc. J'espère que tu trouves cela utile.