Thread-Pooling in C ++ 11

130

Relevante Fragen :

Über C ++ 11:

Über Boost:


Wie erhalte ich einen Pool von Threads, an die Aufgaben gesendet werden können, ohne sie immer wieder zu erstellen und zu löschen? Dies bedeutet, dass persistente Threads ohne Verknüpfung neu synchronisiert werden müssen.


Ich habe Code, der so aussieht:

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

Anstatt Threads bei jeder Iteration zu erstellen und zu verknüpfen, würde ich es vorziehen, Aufgaben bei jeder Iteration an meine Worker-Threads zu senden und sie nur einmal zu erstellen.

Yktula
quelle
1
Hier ist eine verwandte Frage und meine Antwort.
Didierc
1
Sie haben über die Verwendung von TBB nachgedacht (es ist Intel, aber kostenlos und Open Source und macht genau das, was Sie wollen: Sie reichen einfach (rekursiv teilbare) Aufgaben ein und sorgen sich nicht um die Threads)?
Walter
2
Dieses FOSS-Projekt ist mein Versuch, eine Thread-Pool-Bibliothek zu erstellen. Probieren Sie es aus, wenn Sie möchten. -> code.google.com/p/threadpool11
Etherealone
Was ist falsch an der Verwendung von TBB?
Walter

Antworten:

84

Sie können die C ++ - Thread-Pool-Bibliothek unter https://github.com/vit-vit/ctpl verwenden .

Dann kann der von Ihnen geschriebene Code durch den folgenden ersetzt werden

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

Sie erhalten die gewünschte Anzahl von Threads und werden diese bei den Iterationen nicht immer wieder erstellen und löschen.

vit-vit
quelle
11
Dies sollte die Antwort sein; Lesbare, unkomplizierte, übersichtliche und standardkonforme C ++ 11-Bibliothek mit einem Header. Gute Arbeit!
Jonathan H
@ vit-vit kannst du bitte ein beispiel mit einer funktion geben? Wie schieben Sie eine Klassenmitgliedsfunktion beiresults[j] = p.push([&arr, j](int){ arr[j] +=2; });
Hani Goc
1
@HaniGoc Erfassen Sie einfach die Instanz als Referenz.
Jonathan H
@ vit-vit Sie haben eine Pull-Anfrage zur Verbesserung der STL-Version erhalten.
Jonathan H
@ vit-vit: Es ist schwierig, den Betreuer dieser Bibliothek mit Fragen zu kontaktieren.
Einpoklum
82

Dies wird von meiner Antwort auf einen anderen sehr ähnlichen Beitrag kopiert, hoffe es kann helfen:

1) Beginnen Sie mit der maximalen Anzahl von Threads, die ein System unterstützen kann:

int Num_Threads =  thread::hardware_concurrency();

2) Für eine effiziente Threadpool-Implementierung ist es besser, nach dem Erstellen von Threads gemäß Num_Threads keine neuen zu erstellen oder alte zu zerstören (durch Verbinden). Dies führt zu Leistungseinbußen und kann sogar dazu führen, dass Ihre Anwendung langsamer als die serielle Version wird.

Jeder C ++ 11-Thread sollte in seiner Funktion mit einer Endlosschleife ausgeführt werden und ständig darauf warten, dass neue Aufgaben erfasst und ausgeführt werden.

So fügen Sie eine solche Funktion dem Thread-Pool hinzu:

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) Die Infinite_loop_function

Dies ist eine "while (true)" - Schleife, die auf die Task-Warteschlange wartet

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) Erstellen Sie eine Funktion, um Ihrer Warteschlange einen Job hinzuzufügen

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

5) Binden Sie eine beliebige Funktion an Ihre Warteschlange

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

Sobald Sie diese Zutaten integriert haben, haben Sie Ihren eigenen dynamischen Threading-Pool. Diese Threads werden immer ausgeführt und warten auf die Ausführung des Jobs.

Ich entschuldige mich, wenn es einige Syntaxfehler gibt, ich habe diesen Code eingegeben und ich habe ein schlechtes Gedächtnis. Es tut mir leid, dass ich Ihnen nicht den vollständigen Thread-Pool-Code zur Verfügung stellen kann, der meine Jobintegrität verletzen würde.

Bearbeiten: Um den Pool zu beenden, rufen Sie die Methode shutdown () auf:

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() 
}
PhD AP EcE
quelle
Wie haben Sie einen Vektor <Thread>, wenn thread (const thread &) = delete?
Christopher Pisz
1
@ChristopherPisz std::vectorerfordert nicht, dass seine Elemente kopierbar sind. Sie können Vektoren mit Nur-Verschieben-Typen verwenden ( unique_ptr,thread , future, etc.).
Daniel Langr
Wie stoppen Sie in Ihrem obigen Beispiel den Pool? Sollte der condition.waitauch nach einer Variablen suchen stop_und prüfen if (stop_ == true) { break;}?
John
@ John, siehe die Abschaltmethode oben.
PhD AP EcE
2
In shutdown () sollte es thread_vector.clear () sein; anstelle von thread_vector.empty (); Richtig?
Sudheerbb
63

Ein Pool von Threads bedeutet, dass alle Ihre Threads ständig ausgeführt werden - mit anderen Worten, die Thread-Funktion wird nie zurückgegeben. Um den Threads eine sinnvolle Aufgabe zu geben, müssen Sie ein System für die Kommunikation zwischen Threads entwerfen, um dem Thread mitzuteilen, dass etwas zu tun ist, und um die tatsächlichen Arbeitsdaten zu kommunizieren.

In der Regel handelt es sich dabei um eine gleichzeitige Datenstruktur, und jeder Thread befindet sich vermutlich in einer Bedingungsvariablen, die benachrichtigt wird, wenn noch etwas zu tun ist. Nach Erhalt der Benachrichtigung werden einer oder mehrere der Threads aktiviert, stellen eine Aufgabe aus der gleichzeitigen Datenstruktur wieder her, verarbeiten sie und speichern das Ergebnis auf analoge Weise.

Der Thread würde dann prüfen, ob noch mehr Arbeit zu erledigen ist, und wenn nicht, wieder einschlafen.

Das Ergebnis ist, dass Sie dies alles selbst entwerfen müssen, da es keinen natürlichen Begriff von "Arbeit" gibt, der universell anwendbar ist. Es ist ziemlich viel Arbeit und es gibt einige subtile Probleme, die Sie richtig stellen müssen. (Sie können in Go programmieren, wenn Sie ein System mögen, das sich hinter den Kulissen um die Thread-Verwaltung kümmert.)

Kerrek SB
quelle
11
"das alles musst du selbst entwerfen" <- das versuche ich zu vermeiden. Goroutinen scheinen jedoch fantastisch.
Yktula
2
@ Yktula: Nun, es ist eine höchst nicht triviale Aufgabe. Aus Ihrem Beitrag geht nicht einmal hervor, welche Art von Arbeit Sie erledigen möchten, und das ist für die Lösung von grundlegender Bedeutung. Sie können Go in C ++ implementieren, aber es wird eine sehr spezifische Sache sein, und die Hälfte der Leute würde sich beschweren, dass sie etwas anderes wollen würden.
Kerrek SB
19

Ein Threadpool ist im Kern eine Reihe von Threads, die alle an eine Funktion gebunden sind, die als Ereignisschleife fungiert. Diese Threads warten endlos auf die Ausführung einer Aufgabe oder ihre eigene Beendigung.

Der Threadpool-Job soll eine Schnittstelle zum Übermitteln von Jobs, zum Definieren (und möglicherweise Ändern) der Richtlinie zum Ausführen dieser Jobs (Planungsregeln, Thread-Instanziierung, Größe des Pools) und zum Überwachen des Status der Threads und der zugehörigen Ressourcen bereitstellen.

Für einen vielseitigen Pool muss zunächst definiert werden, was eine Aufgabe ist, wie sie gestartet, unterbrochen, was das Ergebnis ist (siehe den Begriff des Versprechens und der Zukunft für diese Frage) und welche Art von Ereignissen die Threads beantworten müssen zu, wie sie damit umgehen, wie diese Ereignisse von denen zu unterscheiden sind, die von den Aufgaben behandelt werden. Wie Sie sehen, kann dies ziemlich kompliziert werden und die Funktionsweise der Threads einschränken, da die Lösung immer komplexer wird.

Das aktuelle Tool für die Behandlung von Ereignissen ist ziemlich barebones (*): Grundelemente wie Mutexe, Bedingungsvariablen und einige Abstraktionen (Sperren, Barrieren). In einigen Fällen können sich diese Abstrationen jedoch als ungeeignet herausstellen (siehe diese verwandte Frage ), und man muss wieder die Grundelemente verwenden.

Andere Probleme müssen ebenfalls gelöst werden:

  • Signal
  • ich / o
  • Hardware (Prozessoraffinität, heterogenes Setup)

Wie würden sich diese in Ihrer Umgebung auswirken?

Diese Antwort auf eine ähnliche Frage verweist auf eine bestehende Implementierung, die für Boost und STL gedacht ist.

Ich habe eine sehr grobe Implementierung eines Threadpools für eine andere Frage angeboten, die nicht viele der oben beschriebenen Probleme anspricht. Vielleicht möchten Sie darauf aufbauen. Vielleicht möchten Sie auch einen Blick auf vorhandene Frameworks in anderen Sprachen werfen, um Inspiration zu finden.


(*) Ich sehe das nicht als Problem, ganz im Gegenteil. Ich denke, es ist der Geist von C ++, der von C geerbt wurde.

didierc
quelle
4
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();
    }
}
pio
quelle
2
Vielen Dank! Dies hat mir wirklich geholfen, mit parallelen Threading-Vorgängen zu beginnen. Am Ende habe ich eine leicht modifizierte Version Ihrer Implementierung verwendet.
Robbie Capps
3

So etwas könnte helfen (aus einer funktionierenden App).

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

Sie können es so verwenden:

thread_pool pool(2);

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

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

Beachten Sie, dass die Neuerfindung eines effizienten asynchronen Warteschlangenmechanismus nicht trivial ist.

Boost :: asio :: io_service ist eine sehr effiziente Implementierung oder eine Sammlung plattformspezifischer Wrapper (z. B. umschließt E / A-Abschlussports unter Windows).

Rustyx
quelle
2
Ist mit C ++ 11 so viel Schub nötig? Wäre das nicht std::threadausreichend?
Einpoklum
Es gibt kein Äquivalent stdfür boost::thread_group. boost::thread_groupist eine Sammlung von boost::threadInstanzen. Aber natürlich ist es sehr einfach, durch boost::thread_groupein vectorvon std::threads zu ersetzen .
Rustyx
3

Bearbeiten: Dies erfordert jetzt C ++ 17 und Konzepte. (Ab dem 12.9.16 ist nur g ++ 6.0+ ausreichend.)

Der Abzug von Vorlagen ist jedoch viel genauer, sodass sich die Mühe lohnt, einen neueren Compiler zu erwerben. Ich habe noch keine Funktion gefunden, die explizite Vorlagenargumente erfordert.

Es nimmt jetzt auch jedes geeignete aufrufbare Objekt ( und ist statisch immer noch typsicher !!! ).

Es enthält jetzt auch einen optionalen Green Threading-Prioritäts-Thread-Pool, der dieselbe API verwendet. Diese Klasse ist jedoch nur POSIX. Es verwendet die ucontext_tAPI für das Wechseln von Userspace-Aufgaben.


Ich habe dafür eine einfache Bibliothek erstellt. Ein Anwendungsbeispiel ist unten angegeben. (Ich beantworte dies, weil es eines der Dinge war, die ich gefunden habe, bevor ich beschlossen habe, es selbst zu schreiben.)

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

Sie können asyncjede Funktion mit einem beliebigen (oder ungültigen) Rückgabewert und beliebigen (oder keinen) Argumenten übergeben, und es wird ein entsprechender zurückgegeben std::future. Um das Ergebnis zu erhalten (oder einfach zu warten, bis eine Aufgabe abgeschlossen ist), rufen Sie get()die Zukunft an.

Hier ist der Github: https://github.com/Tyler-Hardin/thread_pool .

Tyler
quelle
1
Sieht toll aus, wäre aber toll, einen Vergleich mit dem Kopfball von vit-vit zu haben!
Jonathan H
1
@ Sh3ljohn, auf den ersten Blick scheint es, dass sie in der API im Grunde gleich sind. vit-vit verwendet die lockfree-Warteschlange von boost, die besser ist als meine. (Mein Ziel war es jedoch speziell, dies nur mit std :: * zu tun. Ich nehme an, ich könnte die wartungsfreie Warteschlange selbst implementieren, aber das klingt schwierig und fehleranfällig.) Außerdem ist vit-vit's keine .cpp zugeordnet, die ist einfacher für Leute, die nicht wissen, was sie tun. (ZB github.com/Tyler-Hardin/thread_pool/issues/1 )
Tyler
Er / sie hat auch eine Nur-STL-Lösung, die ich in den letzten Stunden gegabelt habe. Zuerst sah sie komplizierter aus als Ihre, mit überall verwendeten Zeigern, aber dies ist tatsächlich erforderlich, um die Größenänderung im laufenden Betrieb richtig zu handhaben.
Jonathan H
@ Sh3ljohn, ah, ich habe die heiße Größenänderung nicht bemerkt. Das ist schön. Ich habe mich entschieden, mir darüber keine Sorgen zu machen, weil es nicht wirklich im beabsichtigten Anwendungsfall liegt. (Ich kann mir keinen Fall vorstellen, in dem ich die Größe persönlich ändern möchte, aber das könnte an mangelnder Vorstellungskraft liegen.)
Tyler
1
Beispiel für einen Anwendungsfall: Sie führen eine RESTful-API auf einem Server aus und müssen die Ressourcenzuweisung zu Wartungszwecken vorübergehend reduzieren, ohne den Dienst vollständig herunterfahren zu müssen.
Jonathan H
3

Dies ist eine weitere Thread-Pool-Implementierung, die sehr einfach, leicht zu verstehen und zu verwenden ist, nur die C ++ 11-Standardbibliothek verwendet und für Ihre Zwecke angesehen oder geändert werden kann. Sie sollte ein guter Einstieg sein, wenn Sie mit der Verwendung von Thread beginnen möchten Pools:

https://github.com/progschj/ThreadPool

Halcyon
quelle
3

Sie können thread_pool aus der Boost-Bibliothek verwenden:

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, []() {
      ...
    });
}

Sie können auch Threadpool aus der Open Source-Community verwenden:

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);
}
Amir Fo
quelle
1

Ein Threadpool ohne Abhängigkeiten außerhalb von STL ist durchaus möglich. Ich habe kürzlich eine kleine Threadpool-Bibliothek nur für Header geschrieben , um genau das gleiche Problem zu beheben. Es unterstützt die dynamische Größenänderung des Pools (Ändern der Anzahl der Mitarbeiter zur Laufzeit), Warten, Stoppen, Anhalten, Fortsetzen usw. Ich hoffe, Sie finden es nützlich.

Cantordust
quelle
Sieht so aus, als hätten Sie Ihr Github-Konto gelöscht (oder den Link falsch verstanden). Haben Sie diesen Code an einen anderen Ort verschoben?
RTPax
1
@rtpax Ich habe das Repo verschoben - ich habe die Antwort aktualisiert, um dies widerzuspiegeln.
Cantordust