Implementierung einer Arbeitsstehliste in C/C++?

Question

Ich bin auf der Suche nach einer korrekten Implementierung einer Arbeitsstehliste in C/CPP. Ich habe mich in Google umgeschaut, aber nichts Nützliches gefunden.

Vielleicht ist jemand mit einer guten Open-Source-Implementierung vertraut? (Ich ziehe es vor, den Pseudo-Code aus den Originalarbeiten nicht zu implementieren).

Answer 1

Kein kostenloses Mittagessen geben.

Bitte sehen Sie sich the original work stealing paper an. Dieses Papier ist schwer zu verstehen. Ich weiß, dass Papier eher theoretischen Beweis als Pseudo-Code enthält. Allerdings gibt es einfach keine solche viel mehr einfache Version als TBB. Wenn überhaupt, wird es keine optimale Leistung geben. Das Stehlen von Arbeit kostet etwas Overhead, daher sind Optimierungen und Tricks sehr wichtig. Insbesondere müssen die Warteschlangen Thread-sicher sein. Die Implementierung von hochskalierbaren und wenig komplexen Synchronisationen ist eine Herausforderung.

Ich frage mich wirklich, warum Sie es brauchen. Ich denke, dass richtige Implementierung bedeutet so etwas wie TBB und Cilk. Arbeitsstehlen ist wiederum schwer zu implementieren.

Answer 2

Werfen Sie einen Blick auf Intel Threading Building Blocks.

http://www.threadingbuildingblocks.org/

Answer 3

Ich glaube nicht, JobSwarm verwendet Arbeit Stehlen, aber es ist ein erster Schritt. Mir sind zu diesem Zweck keine anderen Open-Source-Bibliotheken bekannt.

Answer 4

Werden Sie Ihre Arbeitsaufgaben in kleinere Einheiten aufteilen, um die Arbeit zu stehlen?

Answer 5

Die structured_task_group Klasse PPL verwendet eine Arbeitsstehliste für ihre Implementierung. Wenn Sie ein WSQ für Threading benötigen, würde ich das empfehlen.
Wenn Sie tatsächlich nach der Quelle suchen, weiß ich nicht, ob der Code in ppl.h angegeben ist oder ob ein vorkompiliertes Objekt vorhanden ist. Ich werde nachsehen müssen, wenn ich heute Abend nach Hause komme.

Answer 6

weiß nicht, ob dies eine Hilfe für Sie sein würde, aber einen Blick auf this Artikel über AMDs Entwicklernetzwerk haben, seine einfache, aber man sollte etwas Nützliches

Answer 7

Es gibt ein Werkzeug, um es einfach auf eine sehr elegante Art und Weise zu tun. Es ist eine sehr effektive Möglichkeit, Ihr Programm in kürzester Zeit zu parrallelisieren.

Cilk project

HPC Challenge Award

Unser Cilk Eintrag für den HPC Challenge- Klasse 2 Auszeichnung gewann 2006 den Preis für `` beste Kombination aus Eleganz und Leistung '. Der Preis wurde um SC'06 in Tampa am 14. November 2006 gemacht.

Answer 8

Um zu implementieren "Arbeit stehlen" ist nicht schwer in der Theorie. Sie benötigen eine Reihe von Warteschlangen, die Aufgaben enthalten, die funktionieren, indem Sie eine Kombination aus Computing und anderen Aufgaben erstellen, um mehr Arbeit zu erledigen. Und Sie benötigen atomaren Zugriff auf die Warteschlangen, um neu generierte Aufgaben in diese Warteschlangen zu platzieren.Schließlich benötigen Sie eine Prozedur, die jede Task am Ende aufruft, um mehr Arbeit für den Thread zu finden, der die Aufgabe ausgeführt hat. Diese Prozedur muss in Arbeitswarteschlangen suchen, um Arbeit zu finden.

Die meisten solchen Work-Stealing-Systeme machen die Annahme, dass es eine kleine Anzahl von Threads gibt (in der Regel durch echte Prozessorkerne gesichert), und dass es pro Thread genau eine Arbeitswarteschlange gibt. Dann versuchen Sie zuerst, Arbeit aus Ihrer eigenen Warteschlange zu stehlen, und wenn es leer ist, versuchen Sie, von anderen zu stehlen. Was schwierig wird, ist zu wissen, in welche Warteschlangen man schauen muss; sie für die Arbeit seriell zu scannen, ist ziemlich teuer und kann eine große Menge an Konflikten zwischen Threads erzeugen, die nach Arbeit suchen.

Bis jetzt ist dies alles ziemlich generische Zeug mit einer zwei wichtigen Ausnahmen: 1) Schaltkontexte (z. B. Festlegen von Prozessor Kontextregistern wie einem "Stapel") kann nicht in reinem C oder C++ angegeben werden. Sie können dies beheben, indem Sie zustimmen, einen Teil Ihres Pakets in zielplattformspezifischen Maschinencode zu schreiben. 2) Der atomare Zugriff auf die Warteschlangen für einen Multiprozessor kann nicht rein in C oder C++ erfolgen (der Dekker-Algorithmus wird ignoriert), und daher müssen Sie diejenigen codieren, die Assembler wie das X86 LOCK XCH oder Compare and Swap verwenden. Nun, der Code in der Aktualisierung der Warteschlange beteiligt, sobald Sie einen sicheren Zugriff haben, ist nicht sehr komplex, und Sie könnten leicht schreiben, dass in ein paar Zeilen von C.

Allerdings denke ich, dass Sie finden wird, ist der Versuch, solche zu codieren Ein Paket in C und C++ mit gemischtem Assembler ist immer noch ziemlich ineffizient und Sie werden schließlich das gesamte Ding in Assembler codieren. Alles was übrig bleibt, sind C/C++ - kompatible Einstiegspunkte: -}

Ich habe dies für unsere parallele Programmiersprache PARLANSE getan, die die Idee einer beliebig großen Anzahl von parallelen Berechnungen live und interagierend (synchronisierend) zu jedem Zeitpunkt bietet. Es ist hinter den Kulissen auf einem X86 genau mit einem Thread pro CPU implementiert, und die Implementierung ist komplett in Assembler. Der Arbeitsstehlingcode ist wahrscheinlich 1000 Zeilen insgesamt und sein kniffliger Code, weil er im Nichtkonkurrenzfall extrem schnell sein soll.

Der echte Wermutstropfen für C und C++ ist, wenn Sie eine Aufgabe erstellen, die Arbeit darstellt, wie viel Stapelspeicherplatz Sie zuweisen? Serielle C/C++ - Programme vermeiden diese Frage, indem sie einfach riesige Mengen (z. B. 10 MB) linearer Stapel überlagern, und niemand kümmert sich sehr darum, wieviel von diesem Stapelspeicherplatz verschwendet wird. Aber wenn Sie Tausende von Aufgaben erstellen können und sie alle zu einem bestimmten Zeitpunkt live haben, können Sie nicht vernünftigerweise jeweils 10 MB zuweisen. Jetzt müssen Sie entweder statisch bestimmen, wie viel Stack-Speicherplatz eine Aufgabe benötigt (Turing-hard), oder Sie müssen Stack-Chunks zuweisen (z. B. pro Funktionsaufruf), was von den weit verbreiteten C/C++ - Compilern nicht getan wird (zB die, die Sie wahrscheinlich verwenden). Der letzte Ausweg ist, die Aufgabenerstellung zu drosseln, um sie auf einige hundert zu begrenzen, und einige hundert wirklich große Stapel zu den Aufgaben, die live sind, zu multiplexen. Sie können die letzten nicht ausführen, wenn die Tasks den Status "Sperren"/"Aussetzen" aktivieren können, da Sie auf Ihren Schwellenwert stoßen. Sie können das also nur tun, wenn die Tasks nur berechnen. Das scheint eine ziemlich strenge Einschränkung zu sein.

Für PARLANSE haben wir einen Compiler erstellt, der Aktivierungsaufzeichnungen für jeden Funktionsaufruf auf dem Heap zuweist.

Answer 9

Wenn Sie nach einer eigenständigen Workstable Queue-Klassenimplementierung in C++ auf Pthread oder boost :: thread suchen, viel Glück, meines Wissens gibt es keinen.

Wie jedoch andere gesagt haben, haben Cilk, TBB und Microsofts PPL allesamt funktionierende Implementierungen unter der Haube.

Die Frage ist, wollen Sie eine Workstations Queue verwenden oder implementieren? Wenn Sie nur einen verwenden möchten, dann sind die oben genannten Möglichkeiten gute Ausgangspunkte, die Planung einer 'Aufgabe' in jedem von ihnen reicht aus.

Da BlueRaja sagte, dass die task_group & structured_task_group in PPL dies tun wird, auch beachten Sie, dass diese Klassen in der neuesten Version von Intels TBB ebenfalls verfügbar sind. Die Parallelschleifen (parallel_for, parallel_for_each) werden ebenfalls mit Workstealing implementiert.

Wenn Sie sich Quelltext anschauen und keine Implementierung verwenden müssen, ist TBB OpenSource und Microsoft liefert die Quellen für seine CRT, damit Sie Spelunking machen können.

Sie können auch auf Joe Duffys Blog nach einer C# -Implementierung suchen (aber es ist C# und das Speichermodell ist anders).

-Rick

Answer 10

Die nächste Durchführung dieser Arbeit stehlen Algorithmus ich gefunden habe, ist etwas Wool von Karl-Filip Faxén genannt. src/report/comparison

Answer 11

Ich habe this C project zu C++ portiert.

Das Original Steal kann zu einem fehlerhaften Lesevorgang führen, wenn das Array erweitert wird. Ich habe versucht, den Bug zu beheben, gab aber schließlich nach, weil ich eigentlich keinen dynamisch wachsenden Stack brauchte. Anstatt zu versuchen, Speicherplatz zuzuordnen, gibt die Methode Push einfach false zurück. Der Anrufer kann dann eine Drehwarte, d. H. while(!stack->Push(value)){}, ausführen.

#pragma once 
#include <atomic> 

    // A lock-free stack. 
    // Push = single producer 
    // Pop = single consumer (same thread as push) 
    // Steal = multiple consumer 

    // All methods, including Push, may fail. Re-issue the request 
    // if that occurs (spinwait). 

    template<class T, size_t capacity = 131072> 
    class WorkStealingStack { 

    public: 
    inline WorkStealingStack() { 
     _top = 1; 
     _bottom = 1; 
    } 

    WorkStealingStack(const WorkStealingStack&) = delete; 

    inline ~WorkStealingStack() 
    { 

    } 

    // Single producer 
    inline bool Push(const T& item) { 
     auto oldtop = _top.load(std::memory_order_relaxed); 
     auto oldbottom = _bottom.load(std::memory_order_relaxed); 
     auto numtasks = oldbottom - oldtop; 

     if (
     oldbottom > oldtop && // size_t is unsigned, validate the result is positive 
     numtasks >= capacity - 1) { 
     // The caller can decide what to do, they will probably spinwait. 
     return false; 
     } 

     _values[oldbottom % capacity].store(item, std::memory_order_relaxed); 
     _bottom.fetch_add(1, std::memory_order_release); 
     return true; 
    } 

    // Single consumer 
    inline bool Pop(T& result) { 

     size_t oldtop, oldbottom, newtop, newbottom, ot; 

     oldbottom = _bottom.fetch_sub(1, std::memory_order_release); 
     ot = oldtop = _top.load(std::memory_order_acquire); 
     newtop = oldtop + 1; 
     newbottom = oldbottom - 1; 

     // Bottom has wrapped around. 
     if (oldbottom < oldtop) { 
     _bottom.store(oldtop, std::memory_order_relaxed); 
     return false; 
     } 

     // The queue is empty. 
     if (oldbottom == oldtop) { 
     _bottom.fetch_add(1, std::memory_order_release); 
     return false; 
     } 

     // Make sure that we are not contending for the item. 
     if (newbottom == oldtop) { 
     auto ret = _values[newbottom % capacity].load(std::memory_order_relaxed); 
     if (!_top.compare_exchange_strong(oldtop, newtop, std::memory_order_acquire)) { 
      _bottom.fetch_add(1, std::memory_order_release); 
      return false; 
     } 
     else { 
      result = ret; 
      _bottom.store(newtop, std::memory_order_release); 
      return true; 
     } 
     } 

     // It's uncontended. 
     result = _values[newbottom % capacity].load(std::memory_order_acquire); 
     return true; 
    } 

    // Multiple consumer. 
    inline bool Steal(T& result) { 
     size_t oldtop, newtop, oldbottom; 

     oldtop = _top.load(std::memory_order_acquire); 
     oldbottom = _bottom.load(std::memory_order_relaxed); 
     newtop = oldtop + 1; 

     if (oldbottom <= oldtop) 
     return false; 

     // Make sure that we are not contending for the item. 
     if (!_top.compare_exchange_strong(oldtop, newtop, std::memory_order_acquire)) { 
     return false; 
     } 

     result = _values[oldtop % capacity].load(std::memory_order_relaxed); 
     return true; 
    } 

    private: 

    // Circular array 
    std::atomic<T> _values[capacity]; 
    std::atomic<size_t> _top; // queue 
    std::atomic<size_t> _bottom; // stack 
    }; 

Full Gist (including unit tests). Ich habe nur die Tests auf eine starke Architektur laufen (x86/64), so weit als schwache Architekturen gehen die Leistung kann variieren, wenn Sie versuchen, dies zu verwenden, um auf z.B. Neon/PPC.

Answer 12

OpenMP kann sehr gut arbeits Stehlen obwohl seine genannte rekursive Parallelität unterstützen

OpenMP forum post

Die OpenMP Spezifikation definiert Tasking-Konstrukte (die verschachtelt werden können, so sind sehr geeignet für die rekursive Parallelität) aber tut nicht die Details angeben, wie sie implementiert werden. OpenMP-Implementierungen, einschließlich gcc, verwenden in der Regel eine Art Arbeitsstehlen für Aufgaben, obwohl der genaue Algorithmus (und die resultierende Leistung) variieren können!

Siehe #pragma omp task und #pragma omp taskwait

aktualisieren

Kapitel 9 des Buches C++ Concurrency in Action beschreibt, wie "Arbeit zu stehlen für Poolthreads" implementieren. Ich habe es selbst nicht gelesen/implementiert, aber es sieht nicht so schwierig aus.