threading Problem

[knivil]

Um nochmal klarzustellen: Ich habe Templatest verwendet, um mir Schreibarbeit bei den Typen/Iteratoren zu sparen. Das ganze wurde fuer std::vector<int> gebastelt und getestet. Bei anderen gibt es wahrscheinlich Fehlermeldungen. Auch habe ich mich am Orginalprogramm orientiert.

[treader]

Ich sehe beim urspruenglichen Code zwei Probleme:

1. In accumulate wird unsynchronisiert von mehreren threads auf den nicht thread sicheren std::vector zugegriffen. Da dort die Iteratoren unsynchronisiert modifiziert werden, kann es dort zu Fehlern kommen. Warhscheinlich ist das der Grund fuer den Absturz, aber genau findest Du das nur mit einem Debugger heraus.

2. threadpool::run() ist so implementiert, dass wenn der Pool voll ist eine lang laufende Methode, die dann eingespeist wird, den Pool blockiert, selbst wenn alle pool-threads fertig sind. So kannst Du den Pool nur dann nutzen, wenn eine thread des Pools gerade idle ist, ansonsten wird alles serialisiert.

3. In einer multi-threaded Umgebung wird Dir die try_run Methode nicht viel nutzen, da die Information nach dem Aufruf schon veraltet ist.

4. Du gibst schedule() run() und try_run() Referenzen auf ein task-Objekt, das aber ggf. nicht mehr existiert. Das macht die Nutzung Deines Pools fehleranfaellig.

[SeppJ]

treader schrieb:

1. In accumulate wird unsynchronisiert von mehreren threads auf den nicht thread sicheren std::vector zugegriffen. Da dort die Iteratoren unsynchronisiert modifiziert werden, kann es dort zu Fehlern kommen. Warhscheinlich ist das der Grund fuer den Absturz, aber genau findest Du das nur mit einem Debugger heraus.

Das wird aber vom Standard garantiert, dass dies möglich ist. Es ist absolut ok, gleichzeitig lesend auf das selber Element zuzugreifen. Aber er tut dies noch nicht einmal, da jedes Element genau einmal von einem der Threads angefasst wird. Die Iteratoren darf er ebenfalls ändern wie er lustig ist, diese sind schließlich threadlokal.

Ich kann zwar nicht genau sagen, was der Fehler ist (da er bei mir nicht auftritt), aber dies ist es nicht.

[__Stefan__]

treader schrieb:

Ich sehe beim urspruenglichen Code zwei Probleme:

2. threadpool::run() ist so implementiert, dass wenn der Pool voll ist eine lang laufende Methode, die dann eingespeist wird, den Pool blockiert, selbst wenn alle pool-threads fertig sind. So kannst Du den Pool nur dann nutzen, wenn eine thread des Pools gerade idle ist, ansonsten wird alles serialisiert.

Die Idee von run() ist, dass der Task im this_thread ausgeführt wird, wenn kein Worker frei ist. Der Thread soll arbeiten, nicht warten.
Edit: Bei einem rekursiven Algorithmus könnte das warten den Algorithmus blockieren.
Alternativ kann man machen

if(!pool.try_run(task)) task.run();

was genau das gleiche ist.

Der Mutex dort wird übrigens vorher geunlockt, da blockiert nichts während der ausführung des Tasks.

treader schrieb:

3. In einer multi-threaded Umgebung wird Dir die try_run Methode nicht viel nutzen, da die Information nach dem Aufruf schon veraltet ist.

Die Methode ist eher eine Alternative zu run(), siehe obiges Beispiel.

treader schrieb:

4. Du gibst schedule() run() und try_run() Referenzen auf ein task-Objekt, das aber ggf. nicht mehr existiert. Das macht die Nutzung Deines Pools fehleranfaellig.

Der Task muss natürlich bis nach der Abarbeitung existieren:

task t1(...);
  pool.schedule(t1);
  task t2(...);
  pool.schedule(t2);
  t1.wait();
  t2.wait();

Mittlerweile verwende ich sowieso Futures und übergebe dem Pool packaged_task per move, dann kann man die Tasks auch vergessen wenn sie dem Pool übergeben wurden.

[treader]

Stefan schrieb:

Die Idee von run() ist, dass der Task im this_thread ausgeführt wird, wenn kein Worker frei ist. Der Thread soll arbeiten, nicht warten.
Edit: Bei einem rekursiven Algorithmus könnte das warten den Algorithmus blockieren.
Alternativ kann man machen
C++:
if(!pool.try_run(task)) task.run();

was genau das gleiche ist.

Der Mutex dort wird übrigens vorher geunlockt, da blockiert nichts während der ausführung des Tasks.

Ich versuche es mal mit einem einfachen Beispiel: Du hast einen threadpool mit vier threads. Jetzt speist Du vier tasks ein, die jeder eine Sekunde laufen. 100ms spaeter rufst du run() mit einer task auf, die eine Stunde laeuft. Nach 900ms laeuft dann nur noch die letzte task (direkt in run() im Hauptthread) und Du kannst keine weiteren tasks mehr einspeisen bis die run-Methode zurueck kehrt. Es sei denn, Du hast in einem anderen Thread Deinen Pool ebenfalls bekannt gemacht.

Besser ist es, immer den task in die queue zu stellen, so kannst Du sicher sein, dass jeder frei werdende task das naechste Element bearbeitet, und vom Hauptthread (oder anderen, die den pool nutzen) kannst Du immer etwas in den pool zur Bearbeitung einstellen

[__Stefan__]

treader schrieb:

Besser ist es, immer den task in die queue zu stellen, so kannst Du sicher sein, dass jeder frei werdende task das naechste Element bearbeitet, und vom Hauptthread (oder anderen, die den pool nutzen) kannst Du immer etwas in den pool zur Bearbeitung einstellen

Das ist sicher wünschenswert, aber was machst du, wenn du einen Task hast, der wiederum einen Task erzeugt?

void useless(int depth, thread_pool& pool)
{
	if(depth > 0)
	{
		task t(useless, depth - 1, std::ref(pool));
		pool.schedule(t);
		t.wait();
	}
}
int main()
{
  thread_pool pool(8);

  useless(10, pool);

  return 0;
}

Wenn nicht genügend Worker vorhanden sind, blockiert sich das. Wenn du aber folgendes machst, dann läuft es sicher durch:

void useless(int depth, thread_pool& pool)
{
	if(depth > 0)
	{
		task t(useless, depth - 1, std::ref(pool));

                pool.run(t); // erzeugt nur Task wenn Thread frei

		t.wait();
	}
}
int main()
{
  thread_pool pool(8);

  useless(10, pool);

  return 0;
}

[SeppJ]

Stefan schrieb:

Wenn nicht genügend Worker vorhanden sind, blockiert sich das.

Richtig, das blockiert. Dein Ansatz ist jedoch keine gute Lösung, wie threader mit seinem Beispiel erklärte. Ein besserer Ansatz wäre, an dem Punkt "genügend Worker" anzusetzen: Wenn ein Thread blockiert (zum Beispiel, weil er auf Untertasks wartet), dann wird dem Pool ein neuer Thread hinzugefügt, der eine Task abarbeitet. Später, wenn wieder alle Threads arbeiten, wird ein Thread zerstört, so dass du dich wieder der hardwareabhängigen Optimalzahl näherst, bei der pro Kern ein Thread aktiv(!) ist. "aktiv" ist hier das wichtige Stichwort, wir wollen schließlich keinen Kern blockieren, daher der Ansatz, bei Blockierung einen neuen, aktiven Thread zu erzeugen.

Tipps zur Implementierung dieser Technik findest du in Literatur zu Threadpools.

[threader]

Eine Frage habe ich noch zum urspruenglichen Code:

Warum wird in schedule() (und in run() auch) ein notify_all() verwendet?

void schedule(task& t)
    {
        std::unique_lock<std::mutex> lock(m_mutex);

        m_queue.push(&t);
        m_cond.notify_all();
    }

Ich wuerde ein notify_one() an der Stelle einsetzen, da ja nur fuer einen Thread neue Arbeit da ist.

[trader]

threader schrieb:

Eine Frage habe ich noch zum urspruenglichen Code:

Der ursprüngliche Code ist schrecklich. Es wimmelt vor std::lock_guard<std::mutex> , auch an Stellen, an denen es gar nicht nötig wäre. Code mit Mutexen ist nicht multithreaded.

Angenommen, der Ansatz wäre so gewollt (was er nicht ist).

Dann:
- Als queue eine lock-free queue nehmen (boost hat da eine)
- Als m_idle und done ein atomic nehmen
- std::lock_guardstd::mutex kann dann aus allen(!) Funktion

Allerdings sind immer noch gravierende Designfehler drin, vor allem dieses "nur 1 Thread pro CPU" ist ideal.

Wer sich nicht auskennt soll vorhandene Libraries nehmen.

[threader]

trader schrieb:

Der ursprüngliche Code ist schrecklich. Es wimmelt vor std::lock_guardstd::mutex, auch an Stellen, an denen es gar nicht nötig wäre. Code mit Mutexen ist nicht multithreaded.

Das beantwortet leider nicht meine Frage.

trader schrieb:

Wer sich nicht auskennt soll vorhandene Libraries nehmen.

Das mag sein. Aber wer dazu lernen will sollte keine Angst haben, Fragen zu stellen und Fehler zu machen.

[knivil]

Code mit Mutexen ist nicht multithreaded

Das ist Bullshit.

Als queue eine lock-free queue nehmen

Lock-free wird ueberbewertet. Und wie du bereits an der boost-Doku siehst, steht als Note fast ueberall dabei ... wenn Memoryallocator nicht blockiert.

[trader]

knivil schrieb:

Und wie du bereits an der boost-Doku siehst, steht als Note fast ueberall dabei ... wenn Memoryallocator nicht blockiert.

Zufälligerweise ist der Default-Allocator nicht std::allocator sondern eine stateful LockFree-Implementation.

knivil schrieb:

Code mit Mutexen ist nicht multithreaded

Das ist Bullshit.

Ist vielleicht etwas übertrieben, aber gelockter Code skaliert nicht schön auf viele CPUs.

Stefan schrieb:

Wichtig ist dass die Algorithmen skalieren. Eines Tages werden Wir Desktop-CPUs mit 128 Kernen haben. Die Parallelisierung hat natürlich einen Overhead der sich bei wenigen Kernen stärker bemerkbar macht.

Der Originalcode wird vermutlich ab 4 CPUs immer langsamer und ist ab 8 CPUs langsamer als auf 1 CPU -- wegen dem Locking und notify_all.

[knivil]

Ist vielleicht etwas übertrieben, aber gelockter Code skaliert nicht schön auf viele CPUs.

Code mit shared state/Kommunikationsbedarf skaliert nicht schoen auf mehreren CPUs. Wenn ein Thread ein Ergebnis benoetigt, dann muss er drauf warten. Wenn der Kommunuikationsaufwand im Verhaeltnis zur Anzahl der Threads und Arbeitsaufwand konstant bleibt, dann ist das gut/okay.

Kannst gern meine Quick and Dirty Loesung mit dem Threadpool vergleichen.

[hustbaer]

trader schrieb:

knivil schrieb:

Code mit Mutexen ist nicht multithreaded

Das ist Bullshit.

Ist vielleicht etwas übertrieben, aber gelockter Code skaliert nicht schön auf viele CPUs.

Stefan schrieb:

Wichtig ist dass die Algorithmen skalieren. Eines Tages werden Wir Desktop-CPUs mit 128 Kernen haben. Die Parallelisierung hat natürlich einen Overhead der sich bei wenigen Kernen stärker bemerkbar macht.

Der Originalcode wird vermutlich ab 4 CPUs immer langsamer und ist ab 8 CPUs langsamer als auf 1 CPU -- wegen dem Locking und notify_all.

Ach die Aussage ist doch in der Form auch immer noch Schmarrn^4.

Klar, der Code wird nicht gut damit klarkommen wenn man superkleine Tasks verteilen will.

Wenn die Tasks allerdings "gross" genug sind, sagen wir mal Grössenordnung 1ms oder mehr, dann wird das mit ziemlich vielen CPUs noch ziemlich gut skalieren.

ps: das notify_all gehört natürlich weg, da hast du schon Recht. Nur wird es LANGE nicht so bremsen wie du uns hier weismachen willst. Wenn die Tasks, wie im Originalcode duch die "sleeps" angedeutet, wirklich in der Grössenordnung 100ms liegen, dann wird man auf Maschinen die sich irgendjmd. von uns privat leisten könnte vermutlich nichtmal merken ob notify_all oder notify_one verwendet wird.

[trader]

hustbaer schrieb:

Wenn die Tasks allerdings "gross" genug sind, sagen wir mal Grössenordnung 1ms oder mehr, dann wird das mit ziemlich vielen CPUs noch ziemlich gut skalieren.

Nein, das ist kein Geflame, da steckt eine mathematische Überlegung dahinter:

knivil schrieb:

Kannst gern meine Quick and Dirty Loesung mit dem Threadpool vergleichen.

Annahme: Der Computer hat unendlich viele CPUs.
Knivils Lösung läuft dann in O(log n) Zeit, weil die Rekursion einen schönen Binärbaum erzeugt und die Zeit gerade die Tiefe ist.
Stefans Lösung läuft in O(n), falls der Task eine feste Grösse hat. Einfach weil pro Task 1x gelockt wird.
Haben die Tasks eine Grösse von $\sqrt{n}$, ist die Zeit "nur" $O(\sqrt{n})$ (Wurzel n mal locken, Wurzel n für die Ausführung: deshalb ist diese Grösse optimal)

Heutige Rechner haben nicht unendlich viele CPUs. Mit 2 CPUs ist es gut möglich, dass Stefans Ansatz besser ist. Aber ab irgend einer Grenze ist der nicht lockzentrierte Ansatz von Knivil überlegen. Vielleicht erst ab 128 CPUs falls std::async naiv implementiert ist. Vielleicht schon mit 2 CPUs falls std::async einen Threadpool verwendet. Aber ganz sicher skaliert er besser.

[trader]

Blödes Smilie, keine Ahnung wie das da reingerutscht ist.

[threader]

trader schrieb:

Annahme: Der Computer hat unendlich viele CPUs.
Knivils Lösung läuft dann in O(log n) Zeit, weil die Rekursion einen schönen Binärbaum erzeugt und die Zeit gerade die Tiefe ist.
Stefans Lösung läuft in O(n), falls der Task eine feste Grösse hat. Einfach weil pro Task 1x gelockt wird.
Haben die Tasks eine Grösse von n−√, ist die Zeit "nur" O(n−√) (Wurzel n mal locken, Wurzel n für die Ausführung: deshalb ist diese Grösse optimal)

M.E. sind beide Loesungen O(n), da alle Zahlen aufaddiert werden. Dass da ein Binaerbaum entsteht, hat nur Einfluss auf die Rekursionstiefe, bzw. die Anzahl Threads. Die ist O(ld n).
Ob nun gelockt wird oder nicht, hat auf die asymptotische Komplexitaet keinen Einfluss. Nur der absolute Overhead veraendert sich.

Im Uebrigen ist das rekursive Starten von Threads gerade kein, oder zumindest ein sehr unguenstiger, Anwendungsfall fuer einen Threadpool.

[SeppJ]

threader schrieb:

M.E. sind beide Loesungen O(n), da alle Zahlen aufaddiert werden.

In absoluter CPU-Zeit vielleicht, aber ganz bestimmt nicht in Echtzeit! Sonst wäre jede Art von Parallelisierung vollkommen sinnlos.

[threader]

In absoluter CPU-Zeit vielleicht, aber ganz bestimmt nicht in Echtzeit! Sonst wäre jede Art von Parallelisierung vollkommen sinnlos.

Ob ich parallelisiere oder nicht hat auf die asymptotische Komplelixtaet erst mal keinen Einfluss. Die sagt ja nichts ueber den absoluten Zeitbedarf aus, sondern darueber, wie sich der Zeitbedarf mit der Problemgroesse entwickelt.

O(n) bleibt - bei gleichem Algorithmus - O(n). Die Ausfuehrungszeit reduziert sich durch das parallele Verarbeiten der Daten (jede CPU berechnet nur einen Teil der Daten, aber mit demselben Algorithmus).

Oder platt gesagt:
1000 Werte aufaddieren sind immer 999 Additionen.

[threader]

Ergaenzung:

Bei 4 CPUs waere O(n) dann O(n/4), was aber wieder als O(n) geschrieben wird.