Mise en commun des threads en C++11

Question

Questions pertinentes :

À propos de C++11 :

• C++11 : std::thread pooled ?
• Est-ce que async(launch::async) en C++11 rendra les pools de threads obsolètes pour éviter la création coûteuse de threads ?

A propos de Boost :

• C++ boost thread réutilisation des threads
• boost::thread et en créer un pool !

---

Comment puis-je obtenir un pool de fils a envoyer des tâches à sans les créer et les supprimer à chaque fois ? Cela signifie que les threads persistants peuvent se resynchroniser sans se joindre.

---

J'ai un 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érerais envoyer des tâches à mes threads de travail à chaque itération et ne les créer qu'une seule fois.

Answer 1

Vous pouvez utiliser thread_pool de la bibliothèque de 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 pool de threads 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);
}

Answer 2

Quelque chose comme ceci pourrait vous aider (extrait 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 efficace mécanisme de mise en file d'attente asynchrone n'est pas trivial.

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

Answer 3

Edit : Cela nécessite maintenant C++17 et des concepts. (A partir du 9/12/16, seul g++ 6.0+ est suffisant).

La déduction des modèles est beaucoup plus précise grâce à cela, donc cela vaut la peine de se procurer un compilateur plus récent. Je n'ai pas encore trouvé de fonction qui nécessite des arguments de template explicites.

Il prend aussi maintenant n'importe quel objet appelable approprié ( et est toujours statiquement sûr !!! ).

Il comprend également désormais un pool de threads prioritaires de type "green threading" facultatif utilisant la même API. Cette classe n'est cependant que POSIX. Elle utilise l'API ucontext_t API pour la commutation des 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 à cette question parce que c'est 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 async n'importe quelle fonction avec une valeur de retour quelconque (ou nulle) et n'importe quels arguments (ou aucun) et elle renverra une fonction correspondante std::future . Pour obtenir le résultat (ou simplement attendre qu'une tâche soit terminée), vous appelez get() sur l'avenir.

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

Answer 4

Il semble que le threadpool soit un problème/exercice très populaire :-)

J'en ai récemment écrit un en C++ moderne ; il m'appartient et est disponible publiquement ici -. https://github.com/yurir-dev/threadpool

Il prend en charge les valeurs de retour modélisées, l'épinglage du noyau, l'ordonnancement de certaines tâches. Toute l'implémentation se fait dans deux fichiers .h.

Ainsi, la question originale sera quelque chose comme ceci :

#include "tp/threadpool.h"

int arr[5] = { 0 };

concurency::threadPool<void> tp;
tp.start(std::thread::hardware_concurrency());

std::vector<std::future<void>> futures;
for (int i = 0; i < 8; ++i) { // for 8 iterations,
    for (int j = 0; j < 4; ++j) {
        futures.push_back(tp.push([&arr, j]() {
               arr[j] += 2;
            }));
    }
}

// wait until all pushed tasks are finished.
for (auto& f : futures)
    f.get();
// or just tp.end(); // will kill all the threads

arr[4] = *std::min_element(arr, arr + 4);