ThreadPool - Boost.Thread

Tachyon

Du hampelst doch ständig mit boost::asio herum. Wieso benutzt Du nicht einfach boost::asio::deadline_timer ? Der macht quasi genau das, was Du da von Hand nachbaust.

Kóyaánasqatsi

@last post
Nein, ich möchte das gerne so progg0rn. Aber trotzdem danke, werde ich mir auch noch angucken.

@quasar
Du sagtest ja es sollte funktionieren, wenn der Thread solange wartet und erst dann stoppt, wenn eine Variablenänderung erfolgt.

Dann sollte das ja funktionieren:

#include <iostream>
#include <boost/thread.hpp>

static unsigned int a = 1;

void Callback()
{
    while(a <= 1000)
	{
        std::cout << a++ << std::endl;
	}
}

int main()
{
    //Starten eines neuen Threads -> Aufspalten der Codeausführung
    //z.B. in AddTimer
    boost::thread thread(Callback);

    //der alte main-Thread läuft weiter
    //der neue Thread führt die Endlosschleife aus

    //vor dem Ende des Programms, sollte man warten, bis alle Threads terminiert sind.
    //z.B. mit thread.join();
    //thread.join();

    //Der main-Thread verlässt hier die main-Funktion und das Programm
    //terminiert, inklusive aller noch laufenden Threads
    std::cout << "Programm terminiert" << std::endl;
    return 0;
}

Tut's aber nicht. Gibt gar nichts aus

Tachyon

Kóyaánasqatsi schrieb:

//Der main-Thread verlässt hier die main-Funktion und das Programm
    //terminiert, inklusive aller noch laufenden Threads
    std::cout << "Programm terminiert" << std::endl;
    return 0;

Ohne ein thread::join geht das so nicht. Das hat quasar auch eigentlich recht deutlich gemacht.

Tachyon

Mir ist gerade langweilig, daher mal ein kleines Beispiel (ungetestet):

class SuspekteFadenKlasse
{
public:
    SuspekteFadenKlasse(std::size_t fadenZahl);

    void stoppeLaufmaschen();

private:

    void laufmasche(std::size_t id);

    boost::ptr_vector<boost::thread> fadenBecken;
    bool maschenLaufen;
};

SuspekteFadenKlasse::SuspekteFadenKlasse(std::size_t fadenZahl)
: maschenLaufen(true)
{
    for(std::size_t n = 0; n != fadenZahl; ++n)
    {
        fadenBecken.push_back(new boost::thread(boost::bind(&SuspekteFadenKlasse::laufmasche, this, n)));
    }
}

void SuspekteFadenKlasse::stoppeLaufmaschen()
{
    maschenLaufen = false;
    for(std::size_t n = 0; n != fadenBecken.size(); ++n)
    {
        fadenBecken[n].join(); //alle Threads durchlaufen und aufs Ende warten
    }
}

void SuspekteFadenKlasse::laufmasche(std::size_t id)
{
    std::ostringstream amazonasDelta;
    while(maschenLaufen)
    {
        //stringstream sorgt auf den handelsueblichen compilern (MSVC, gcc)
        //dafuer, dass die Ausgabe nicht durcheinander geht
        amazonasDelta.str("");
        amazonasDelta << "Ich bin aus Faden: " << id << '\n';
        std::cout << amazonasDelta.str();
        boost::this_thread::sleep(boost::posix_time::seconds(1));
    }
}

int main()
{
    SuspekteFadenKlasse strumpfhose(5);
    //wuerde jetzt erstmal auch einfach so beenden...
    std::string kommando;
    while(kommando != "quit")
    {
        std::getline(std::cin, kommando); //...aber wir warten.
    }
    strumpfhose.stoppeLaufmaschen(); //hier wird darauf gewartet, dass alle threads terminieren
}

Kóyaánasqatsi

Ich will ein Kind von dir. Mal wieder.

hustbaer

Also da sind streng genommen zwei Fehler drin.

Man darf nicht einfach so mit mehreren Threads gleichzeitig nach std::cout schreiben. Die meisten Compiler crashen nicht, aber garantiert ist nix. k.A. ob der kommende Standard das ändern wird.
Das "cancel flag" ist nicht "synchronisiert". Die Minimallösung wäre es volatile zu machen. Die bessere Lösung wäre es mit einer Mutex zu schützen.

Was die komischen Namen sollen entzieht sich mir im Moment, aber ich mache mal einfach mit

#include <boost/thread.hpp>
#include <boost/ptr_container/ptr_vector.hpp>
#include <boost/bind.hpp>

class SuspekteFadenKlasse 
{ 
public: 
	SuspekteFadenKlasse(std::size_t fadenZahl); 

	void stoppeLaufmaschen(); 

private: 
	void laufmasche(std::size_t id); 

	boost::ptr_vector<boost::thread> fadenBecken;  // müsste mit aktuellen compilern auch mit nem normalen std::vector gehen, move semantik und so

	boost::mutex torWaechter;
	boost::condition_variable_any postillion;
	bool maschenLaufen; 
}; 

SuspekteFadenKlasse::SuspekteFadenKlasse(std::size_t fadenZahl) 
: maschenLaufen(true) 
{ 
	for(std::size_t n = 0; n != fadenZahl; ++n) 
	{ 
		fadenBecken.push_back(new boost::thread(boost::bind(&SuspekteFadenKlasse::laufmasche, this, n))); 
	} 
} 

void SuspekteFadenKlasse::stoppeLaufmaschen() 
{ 
	{
		boost::unique_lock<boost::mutex> torSchluss(torWaechter);
		maschenLaufen = false; 
		postillion.notify_all();
	}

	for(std::size_t n = 0; n != fadenBecken.size(); ++n) 
	{ 
		fadenBecken[n].join(); //alle Threads durchlaufen und aufs Ende warten 
	} 
} 

void SuspekteFadenKlasse::laufmasche(std::size_t id) 
{ 
	boost::unique_lock<boost::mutex> torSchluss(torWaechter);
	while(maschenLaufen)
	{ 
		std::cout << "Ich bin aus Faden: " << id << '\n'; 

		// wenns was zu machen gibt was länger dauert, sollte man während man es macht dem torWaechter frei geben.
		// allerdings sollte der torWaechter NICHT frei haben, während man maschenLaufen abfragt.

		// warten bis es post gibt, allerdings maximal 1 sekunde
		postillion.timed_wait(torWaechter, boost::posix_time::seconds(1));
	} 
} 

int main() 
{ 
	SuspekteFadenKlasse strumpfhose(5); 
	//wuerde jetzt erstmal auch einfach so beenden... 
	std::string kommando; 
	while(kommando != "quit") 
	{ 
		std::getline(std::cin, kommando); //...aber wir warten. 
	} 
	strumpfhose.stoppeLaufmaschen(); //hier wird darauf gewartet, dass alle threads terminieren 
}

nurf

hustbaer schrieb:

Also da sind streng genommen zwei Fehler drin.

Das "cancel flag" ist nicht "synchronisiert". Die Minimallösung wäre es volatile zu machen. Die bessere Lösung wäre es mit einer Mutex zu schützen.

Leider sind in Punkt 2) auch zwei Fehler drin.

1. Volatile ist kein atomic_compare_and_set, womit also Race Conditions genauso wenig ausgeschlossen sind.

2. Ein Mutex ist keine Memory-Fence. Da ein x86 mit out-of-order execution arbeitet, ist ein Write-Back vor dem Mutex-Unlock von dem Flag nicht garantiert.
Warum es aber meistens funktioniert ist:
2.1 Weil die meisten Mutex-Implementierungen eine Memory-Fence verwenden.
2.2 Weil x86er Cache-Kohärent sind.

Wenn ich jedes mal einen Euro kriegen würde, wenn jemand volatile in Zusammenhang mit Threads erwähnt ...

hustbaer

nurf schrieb:

hustbaer schrieb:

Also da sind streng genommen zwei Fehler drin.

Das "cancel flag" ist nicht "synchronisiert". Die Minimallösung wäre es volatile zu machen. Die bessere Lösung wäre es mit einer Mutex zu schützen.

Leider sind in Punkt 2) auch zwei Fehler drin.

1. Volatile ist kein atomic_compare_and_set, womit also Race Conditions genauso wenig ausgeschlossen sind.

volatile ist für ein Cancel-Flag praktisch gesehen ausreichend. Für ein Cancel-Flag brauche ich weder atomare Zugriffe noch Memory-Ordering. Wichtig ist nur dass die Änderung des schreibenden Threads irgendwann mal für andere Threads sichtbar wird, und dass der lesende Thread nie "true" liest, obwohl keiner jemals "true" reingeschrieben hat. Ganz streng genommen garantiert das der C++ Standard nicht, aber ich kenne keine Archtektur wo es nicht funktionieren würde. Auch Alpha etc. sind diesbezüglich problemlos.
Wenn es um mehr als nur ein Cancel-Flag geht, braucht man natürlich andere Mittel.

2. Ein Mutex ist keine Memory-Fence. Da ein x86 mit out-of-order execution arbeitet, ist ein Write-Back vor dem Mutex-Unlock von dem Flag nicht garantiert.
Warum es aber meistens funktioniert ist:
2.1 Weil die meisten Mutex-Implementierungen eine Memory-Fence verwenden.
2.2 Weil x86er Cache-Kohärent sind.

Was eine Mutex genau ist, ist wohl Definitionssache. Ich verwende hier boost::mutex, und boost::mutex garantiert acquire/release Semantik. Egal auf welcher Plattform. PTHREADs garantiert soweit ich weiss das selbe, Windows CRITICAL_SECTIONs und Mutexen ebenso.
Ich sehe hier also nicht mal ganz ganz streng genommen einen Fehler.

Wenn ich jedes mal einen Euro kriegen würde, wenn jemand volatile in Zusammenhang mit Threads erwähnt ...

Wenn ich jedes mal einen Euro kriegen würde, wenn jemand eine sinnlose Bemerkung macht...

nurf

hustbaer schrieb:

volatile ist für ein Cancel-Flag praktisch gesehen ausreichend. Für ein Cancel-Flag brauche ich weder atomare Zugriffe noch Memory-Ordering. Wichtig ist nur dass die Änderung des schreibenden Threads irgendwann mal für andere Threads sichtbar wird, und dass der lesende Thread nie "true" liest, obwohl keiner jemals "true" reingeschrieben hat.

Dann lass das volatile in Zukunft weg und du wirst sehen, es funktioniert trotzdem.

hustbaer

nurf schrieb:

hustbaer schrieb:

volatile ist für ein Cancel-Flag praktisch gesehen ausreichend. Für ein Cancel-Flag brauche ich weder atomare Zugriffe noch Memory-Ordering. Wichtig ist nur dass die Änderung des schreibenden Threads irgendwann mal für andere Threads sichtbar wird, und dass der lesende Thread nie "true" liest, obwohl keiner jemals "true" reingeschrieben hat.

Dann lass das volatile in Zukunft weg und du wirst sehen, es funktioniert trotzdem.

Das tut es nur, wenn der Code in der Schleife für den Compiler zu kompliziert wird, so dass er nichtmehr sicher sein kann, dass das Cancel-Flag nicht irgendwo geändert werden könnnte.

Das funktioniert z.B. nicht:

while (!cancelFlag);

Hier erkennt der Compiler dass cancelFlag in der Schleife nicht verändert wird, und zieht den Test "!cancelFlag" aus der Schleife raus.
volatile verhindert das.

Wieso ich damit pokern sollte dass der Compiler die mögliche (und durch das fehlende volatile erlaubte) Optimierung nicht findet, weiss ich nicht.

nurf

Also entweder verwechselst du da was mit Java oder C#.

http://software.intel.com/en-us/blogs/2007/11/30/volatile-almost-useless-for-multi-threaded-programming/

Soll mir egal sein, im Internet findet sich genug zu der Thematik.

Noch ein Zitat zum Abschluss:

If your multithreaded code works properly with volatile and doesn’t work without, then either your C++ implementation carefully implemented volatile to work with threads (less likely), or you simply got lucky (more likely).

hustbaer

Ich verwechsle nichts, und ich kenne auch die diversen Zitate, Artikel etc.

Volatile in C bzw. C++ zwingt den Compiler dazu, eine load bzw. store Instruction zu erzeugen. Nicht mehr und nicht weniger.

Um den Rest kümmert sich in diesem Fall die CPU, da, wie ich schon geschrieben habe, Speicher-Sichtbarkeit etc. für ein Cancel-Flag vollkommen wurscht sind. Genauso ob der Zugriff atomar erfolgt oder nicht.

Das ist einer der wenigen Spezialfälle, wo man in C bzw. C++ mit volatile, ohne zusätzliche, "freiwillige" Garantien des Compilers, etwas Sinnvolles anstellen kann.

Vorausgesetzt man hat eine CPU, die die Caches *irgendwann mal* selbst updated, ODER ein Betriebssystem welches hin und wieder mal etwas macht was die Caches flusht. Und ich schätze dass das jedes OS welches Threads verwendet tut, nämlich wenn der Scheduler anläuft um zu gucken ob auf einen anderen Thread umgeschaltet werden sollte.

Ich weiss dass es keine vom Standard garantierte Sache ist. Ich sage nur: es funktioniert so-gut-wie überall, wenn nicht überhaupt überall.