3. Threadgestaltung bei erwünschtem Abbruch

Threadgestaltung bei erwünschtem Abbruch


  • Stimmt, Threads sind eigentlich ganz unkompliziert und nicht wirklich schwer zu verstehen.
    Das Problem ist die Komplexität der Probleme die man sich mit diesen simplen Threads schaffen kann 😉

    0

  • Eisflamme schrieb:

    sorry,dann habe ich wohl cvs noch nicht kapiert. ich meine,klar. das volatile fehlt hier,aber das allein würde ja trotzdem nix threadsafe machen.

    Nein, das würde es nicht, dafür ist es ja auch nicht da. Für die Threadsafety wäre die Condition Variable zuständig, die absolut nichts mit dem bool Flag zu tun hat. Wie gesagt, wenn dein Compiler es unterstützt, verwend für das Flag besser einen std::atomic<bool> als volatile.

    0
  • ?

    Und wie genau würde ein std::atomic deiner Meinung nach nun dabei helfen?

    std::atomic garantiert dass der Compiler die nötige Synchronisation sicherstellt.
    Der Compiler weiß, dass der Wert einer Variablen asynchron durch einen anderen Thread verändert werden kann.
    Das bedeutet, dass alle Arten spekulativer Compiler-Optimierungen nicht erlaubt sind.
    Dazu sind die Operationen (lesen und schreiben) atomar, entweder sie sind durchgeführt oder nicht. Keine halben Sachen.

    EDIT: Ok, es könnte natürlich passieren dass auf irgendeiner äußerst exotischen Architektur beim gleichzeitigen Schreiben im einen und Lesen im anderen Thread Datenmüll gelesen wird.
    Selbst dann macht das aber wohl nix.

    Das ist undefined behaviour.

    0

  • "Eigentlich ganz unkompliziert" ist mMn. eine relativ wenig sinnvolle Aussage, wenn man deutlich sehen kann, dass so-gut-wie jeder bei einem bestimmten Thema Fehler macht.

    0

  • einwurf schrieb:

    Und wie genau würde ein std::atomic deiner Meinung nach nun dabei helfen?

    std::atomic garantiert dass der Compiler die nötige Synchronisation sicherstellt.
    Der Compiler weiß, dass der Wert einer Variablen asynchron durch einen anderen Thread verändert werden kann.
    Das bedeutet, dass alle Arten spekulativer Compiler-Optimierungen nicht erlaubt sind.
    Dazu sind die Operationen (lesen und schreiben) atomar, entweder sie sind durchgeführt oder nicht. Keine halben Sachen.

    EDIT: Ok, es könnte natürlich passieren dass auf irgendeiner äußerst exotischen Architektur beim gleichzeitigen Schreiben im einen und Lesen im anderen Thread Datenmüll gelesen wird.
    Selbst dann macht das aber wohl nix.

    Das ist undefined behaviour.

    Ja, wie gleich darauf in meinem Edit geschrieben hast du da völlig recht 😉

    0

  • einwurf schrieb:

    Und wie genau würde ein std::atomic deiner Meinung nach nun dabei helfen?

    std::atomic garantiert dass der Compiler die nötige Synchronisation sicherstellt.
    Der Compiler weiß, dass der Wert einer Variablen asynchron durch einen anderen Thread verändert werden kann.
    Das bedeutet, dass alle Arten spekulativer Compiler-Optimierungen nicht erlaubt sind.
    Dazu sind die Operationen (lesen und schreiben) atomar, entweder sie sind durchgeführt oder nicht. Keine halben Sachen.

    Nein.
    std::atomic garantiert nur atomare Zugriffe + Ordering, viele Optimierungen bleiben nach wie vor erlaubt.
    Ein std::atomic Write muss z.B. nie den Bus sehen (es muss nichtmal Code dafür generiert werden), es sei denn es wird mit volatile kombiniert.

    std::atomic<int> ai;
std::atomic<int> volatile avi;

void foo()
{
    ai = 42; // darf wegoptimiert werden
    ai = 0;

    avi = 42; // darf NICHT wegoptimiert werden
    avi = 0;
}

    @dot:
    Soviel zum Thema unkompliziert, nen 😉

    0

  • @hustbaer: Ich meinte ja nicht dass es einfach ist, im Gegenteil. Ich meine nur dass das Problem nicht die Kompliziertheit einzelner Komponenten (Threads), sondern die Komplexität des sich ergebenden Gesamtsystems ist 😉

    0
  • E

    Ok, jetzt bin ich im Bilde, was CVs sind, aber ich kann mein Problem aus Mangel an Intelligenz leider trotzdem nicht lösen.

    Also ich habe ja nicht Mal eine Queue. Es gibt halt eine aktuell laufende Rechnung und es gibt ein WorkPackage für die nächste Rechnung. Mein erster Versuch war jetzt etwa so:

    void process() // run
{
while(true)
{
mutex.lock();
cv.wait(&mutex);
terminate = false;
currentWorkPackage = nextWorkPackage;
mutex.unlock();
calculate();
}
}

void startCalculation(/* ... */)
{
nextWorkPackage = /* ... */;

terminate = true;

cv.wakeUpAll();
}

    Also für einen einfachen Fall, dass er gerade wartet, klappt es. Für den einfachen Fall, dass er gerade am Arbeiten ist, klappt es auch (keiner wartet -> keiner wird geweckt und terminate = true greift).

    Aber wenn jetzt zwei Mal hintereinander geklickt wird, während er wartet, dann wird der Thread sofort aufgeweckt und setzt erstmal terminate = false (korrekt), aber der zweite Aufruf soll terminate = true setzen... jetzt kann es doch sein, dass der zweite startCalculation schneller ist und dadurch der process-Thread terminate = false noch hinkriegt... also wird dann ja wieder das zweite start ignoriert, oder?

    Edit: Außerdem fällt mir noch auf, dass er ja eine Eingabe ignoriert, wenn sie kurz nach der Berechnung reinkommt, weil da schließlich gerade niemand wartet. 😞

    0

  • dot schrieb:

    @hustbaer: Ich meinte ja nicht dass es einfach ist, im Gegenteil. Ich meine nur dass das Problem nicht die Kompliziertheit einzelner Komponenten (Threads), sondern die Komplexität des sich ergebenden Gesamtsystems ist 😉

    Achso, OK, ja. Das kann man wohl so sagen.
    Wobei das glaub' ich bei vielen Dingen so ist.

    0

  • @Eisflamme
    Willst du dass er immer abbricht, sobald ein neuer Task in die "Queue" gesteckt wurde?

    Dann vielleicht ca. so?

    void Queue::IsEmpty() const
{
    scoped_lock lock(m_mutex);
    return IsEmptyNoLock();
}

void Queue::IsEmptyNoLock() const
{
    return m_nextTask == 0;
}

void Queue::IsAbort() const
{
    scoped_lock lock(m_mutex);
    return IsAbortNoLock();
}

void Queue::IsAbortNoLock() const
{
    return m_abortFlag;
}

Task* Queue::GetTaskNoLock(scoped_lock& lock)
{
    // warten bis ein neuer task da ist ODER abbruchsignal (Queue-Klasse wird zerstört) vorliegt
    while (IsEmptyNoLock() && !IsAbortNoLock())
        m_queueNotEmptyCondition.wait(lock);

    // abort flag checken
    if (IsAbortNoLock())
        return 0;
    else if (IsEmptyNoLock())
        assert(0 && "shit");

    // task "konsumieren" und zurückgeben
    Task* task = m_nextTask;
    m_nextTask = 0;
    return task;
}

Task* Queue::GetTask()
{
    scoped_lock lock(m_mutex);
    return GetTaskNoLock(lock);
}

void Queue::Process()
{
    scoped_lock lock(m_mutex);

    while (!IsAbortNoLock())
    {
        // einen Task abholen ...
        Task* task = GetTaskNoLock(lock);

        // ... und ausführen
        lock.unlock();

        if (task)
            Compute(task);

        lock.lock();
    }
}

void Queue::Compute(Task* task)
{
    for (...)
    {
        // Berechnung abbrechen sobald entweder "abort" gesetzt ist, oder aber ein neuer Task ein der Queue steckt
        // (=die Queue nicht mehr leer ist)
        bool const stopComputation = IsAbort() || !IsEmpty(); // besser wäre ein gemeinsamer getter, sonst muss in IsAbort() und IsEmpty() jedes mal die mutex gelockt und wieder freigegeben werden
        if (stopComputation)
            return;

        // ...
    }
}
    0
  • E

    Genau, bei neuem Task soll die Berechnung abgebrochen werden, weil die Ergebnisse sowieso obsolet wären.

    Mir fällt es noch etwas schwer solchen Code zu lesen, z.B. weiß ich gerade nicht, wofür GetTask genutzt werden soll. Und mich würde natürlich interessieren, wie das Setzen von einem Task geschieht. Geschieht das ohne Mutex und alles? Einfach task setzen und n Wake schießen?

    Und dann zu scoped_lock: Also in Process locked er den Mutex und in IsAbort() wartet er in IsAbort dann ja, bis Freigabe erfolgt, also hängt er erstmal da fest?

    0

  • Das setzen der Tasks musst du natürlich schon mit gelockter Mutex machen.
    Und in Process() wird der Lock ja temporär wieder freigegeben. Sonst würde das ja wirklich alles blockieren.

    Soll auch nicht mehr als eine schnelle Skizze sein wie man sowas implementieren kann.

    Und was GetTask() macht: na so lange warten bis es einen Task abzuarbeiten gibt ODER die Queue zerstört werden soll ("abort"), und den Task (bzw. bei abort eben 0) dann zurückgeben.

    0
  • E

    Ja okay, aber ich meine jetzt - und sorry, wenn ich mich da einfach doof anstelle -, dass doch schon beim Schleifenkopf in GetTaskNoLock eine Sperre entsteht? Ich meine, in Zeile 1 von GetTaskNoLock wird ein scoped_lock erstellt und im Schleifenkopf mit IsAbort() wird ein weiterer scoped_lock in IsAbort() angefasst. Hängt er dann nicht schon da? Oder macht das nichts, weil bei der Taskzuweisung mit dem Wake eben dort geweckt wird?

    Und die Frage auf GetTask meinte, wann das aufgerufen wird vom Benutzer o.ä. Der wird ja nur Compute() aufrufen oder halt den Task ändern.

    Vll. sollte ich mehr rumprobieren, um mich dem Thema anzunähern, mir kommt das unfassbar schwierig vor...

    0

  • Err, hatte da noch nen Kleinen Schönheitsfehler drinnen.
    Task* Queue::GetTaskNoLock(Lock& lock) muss Task* Queue::GetTaskNoLock(scoped_lock& lock) heissen.

    Also...

    scoped_lock ist ne Hilfsklasse die im Konstruktor "Lock" auf die übergebene Mutex macht, und im Destruktor "Unlock".
    Dazwischen kann man selbst manuell "lock" bzw. "unlock" machen -- wobei scoped_lock allerdings immer nur "1x gelockt" sein kann. 2x "Lock" hintereinander ist also ein Fehler und wird mit nem ASSERT() "belohnt", 2x "Unlock" genau so.

    Was die ganzen "XxxNoLock" Funktionen angeht: die heissen so, weil sie selbst keinen Lock holen -- der Aufrufer muss sich darum kümmern. Das ist z.T. ne Performance-Optimierung, bzw. wenn man Condition-Variablen verwendet braucht man es auch öfters -- beim "wait" auf die Condition-Variable darf die Mutex ja nicht rekursiv gelockt sein.

    Queue::GetTaskNoLock() verlangt dabei vom Aufrufer dass eine Referenz auf das zum sperren der Mutex verwendete scoped_lock Objekt mitgegeben wird.
    Diese Referenz wird dann weiters an die wait() Funktion der CV übergeben. Diese wiederrum macht intern ein "unlock-wait-lock". D.h. während man wartet dass die CV signalisiert wird, ist die Mutex frei.

    0
  • E

    Danke für Erklärung und die ganze Geduld!

    Nur der Vollständigkeit halber, das Hinzufügen eines neuen Tasks ginge jetzt so?

    void setTask(Task& task)
{
{
scoped_lock lock(m_mutex);
m_nexttask = task;
}
m_queueNotEmptyCondition.notify_one();
}

    ?

    Oh und - ich komme mit dem Verständnis so langsam weiter -, wie ist das denn am Ende der Schleife in Process: Der Mutex wird gelockt, dann kommt aber IsAbort im Schleifenkopf und er hängt dort. Aber soll der dort hängen? Wäre da nicht ein IsAbortNoLock() sinnvoller?

    0

  • Eisflamme schrieb:

    Danke für Erklärung und die ganze Geduld!

    Nur der Vollständigkeit halber, das Hinzufügen eines neuen Tasks ginge jetzt so?

    void setTask(Task& task)
{
{
scoped_lock lock(m_mutex);
m_nexttask = task;
}
m_queueNotEmptyCondition.notify_one();
}

    ?

    Ja, z.B.

    Also mal abgesehen von den ganzen Dingen die mit Threading nix zu tun haben, wie dass irgendwer die Tasks wieder freigeben muss, dass meine "Queue" gar keine Queue ist (weil sie immer nur einen "next task" beherrbergen kann) etc.
    Ich persönlich mache das signal() immer während die Mutex noch gelockt ist, aber das ist AFAIK nicht gefordert. Und ich würde hier notify_all machen, auch hauptsächlich aus Gewohnheit, und weil ja theoretisch ein zweiter Worker-Thread dazukommen könnte.
    EDIT: OK, das mit "Thread dazukommen" ist vermutlich Unsinn. Reicht ja wenn einer aufwacht. Trotzdem, so lange ich keinen besonders gute Grund sehe mit notify_one() zu arbeiten mache ich immer notify_all(). Spart mit Kopfschmerzen während ich (nicht) darüber nachdenken muss ob notify_one() auch sicher nicht dazu führen kann dass irgendwo irgendwie kein Thread mehr weitermacht, weil sich keiner zuständig gefühlt hat 🙂 /EDIT

    Oh und - ich komme mit dem Verständnis so langsam weiter -, wie ist das denn am Ende der Schleife in Process: Der Mutex wird gelockt, dann kommt aber IsAbort im Schleifenkopf und er hängt dort. Aber soll der dort hängen?

    Nö, soll er nicht. Muss er auch nicht, wenn die Mutex "rekursiv" ist. Also sich mehrfach vom selben Thread aus locken lässt. Was jetzt Vor- und Nachteile hat. (Vorteil: man muss weniger aufpassen. Nachteil: man übersieht schneller Fälle wo eine "doppelt" gelockte Mutex an das wait() einer CV übergeben wird, was dann meist böse endet)

    Wäre da nicht ein IsAbortNoLock() sinnvoller?

    Auf jeden Fall, hab' ich übersehen 🙂
    Ich korrigier das dann mal schnell...

    ps:

    Um Fehler zu vermeiden, bietet es sich an a) wenn's leicht geht eben nicht mit rekursiven Mutexen zu arbeiten und b) allen XxxNoLock Funktionen eine Referenz auf das scoped_lock Objekt als Parameter zu verpassen.
    Dadurch vermeidet man Unachtsamkeitsfehler, wo man XxxNoLock aufruft ohne die Mutex gelockt zu haben.

    0
  • E

    Okay, damit sind meine aktuellen Fragen beantwortet. Habe noch etwas herumgespielt mit QT- und Boost-Mutexen und habe das Gefühl es jetzt wirklich verstanden zu haben. Vielen Dank für die ganze Mühe 🙂

    0
Anmelden zum Antworten