Threadgestaltung bei erwünschtem Abbruch

knivil

Threads sind halt so ein Thema was kaum wer wirklich versteht.

Threads sind sind keine Quantenphysik. Keine Anhnung, was an Threads so schwer ist zu verstehen.

Es gibt dort aber keine Data Race und ich sehe nicht was ein Atomic dort bringen würde...

bool is_aborted

repräsentiert eine Speicherstelle, auf welche mehrere Threads nicht synchronisiert und nicht atomar zugreifen, wobei ein Thread schreibend. Damit liegt ein Data Race vor.
Da es sich hier um nicht-atomare lesende und schreibende Operationen auf diese Variable handelt, verwendest du volatile, was aber meiner Meinung nach nur ein Notbehelf ist.

Denn:

If you do use volatile remember that its detailed semantics vary dramatically across platforms. On some platforms, two consecutive volatile stores may become visible out of order to two consecutive volatile loads in another thread. On other platforms, that is explicitly prevented. Thus platform-dependent mechanisms for memory ordering are usually also needed. (The atomic_ops package uses volatile in platform-dependent ways internally, but adds fences to enforce requested memory ordering.)

It is also important to remember that volatile updates are not necessarily atomic. They may appear to be carried out piecemeal. Whether or not they actually are atomic depends on the platform and alignment.

dot

einwurf schrieb:

Es gibt dort aber keine Data Race und ich sehe nicht was ein Atomic dort bringen würde...
bool is_aborted
repräsentiert eine Speicherstelle, auf welche mehrere Threads nicht synchronisiert und nicht atomar zugreifen, wobei ein Thread schreibend. Damit liegt ein Data Race vor.
Da es sich hier um nicht-atomare lesende und schreibende Operationen auf diese Variable handelt, verwendest du volatile, was aber meiner Meinung nach nur ein Notbehelf ist.

Und wie genau würde ein std::atomic deiner Meinung nach nun dabei helfen? Es gibt keine zwei Threads die jemals gleichzeitig auf den bool schreiben würden. Ich seh also nicht, dass ein Atomic hier irgendwas ändern würde...

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, wobei es aber vermutlich UB ist. Mit einem atomic wäre dagegen natürlich garantiert dass das auf keinen Fall passiert.
Ich hab mit solchen Dingen meist eher ganz unten wo zu tun, wo es um Performance auf ganz bestimmter Hardware mit ganz bestimmten Compilern und bestimmtem Verhalten geht, da übersieht man sowas schnell mal

Also Ja: Besser std::atomic<bool>, sofern dein Compiler es unterstützt.

hustbaer

knivil schrieb:

Threads sind halt so ein Thema was kaum wer wirklich versteht.

Threads sind sind keine Quantenphysik. Keine Anhnung, was an Threads so schwer ist zu verstehen.

Tjoah.
Entweder du überschätzt dich gewaltig, oder du überschätzt den Rest der Programmierer-Gemeinde gewaltig.
Glaub mir, so viel Verwirrung wie beim Thema Threads, vor allem wenn man mal mit Atomics etc. anfängt, gibt es sonst auf kaum einem Gebiet. Zumindest nicht auf einem, mit dem so viele Leute zu tun haben wie mit Threads.

Wenn du da wirklich anderer Meinung bist, tendiere ich mal zu der Variante wo du dich überschätzt.

Eisflamme

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.

dot

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

dot

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.

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.

hustbaer

"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.

dot

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

hustbaer

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

dot

@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

Eisflamme

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.

hustbaer

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.

hustbaer

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

        // ...
    }
}

Eisflamme

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?

hustbaer

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.

Eisflamme

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...

hustbaer

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.

Eisflamme

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?