Atomic / Lockfree Programmierung

[Kr0e]

Hallo liebe Javagemeinde,

ich habe vor einiger Zeit mal einen Beitrag zum Thema "Lockfree"-Programming gesehen und versuche nun in meinem Projekt die Synchronisierung zu entfernen um mehr Durchsatz und von daher mehr Leistung zu erhalten, da Synchronisierung wohl "den Tod" bei
Multicores bedeutet. Bzw effektiv die Leistungssteigerung (z.B. durch einen Multicore) verhindert/verlangsamt.

Ich hoffe, dass es hier ein paar Experten zu diesem Thema gibt, da es meiner Meinung nach ein ziemlich komplexes Thema ist:

Ich habe einer Art Eventklasse, bei der man "Runnable"s registrieren kann, die bei Eintritt des Events benachrichtigt werden.
Ist das Event bereits eingetreten, soll diese "Registrierung" dafür sorgen, dass das Runnable SOFORT ausgeführt wird.

Mit synchronized kein Ding aber natürlich dickes Bottleneck, da ich diese Klasse SEHR OFT in meinem Projekt verwende!
Ich schreibe ein asynchrones Messaging-System, ähnlich JMS aber etwas abstrakter und besser einsetzbar.

Hier ist eine stark vereinfachte Version dieser Klasse:

package foxnet.system.util.concurrent.event;

import java.util.Queue;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;

public class Event {

    private final Queue<Runnable> queue = new ConcurrentLinkedQueue<Runnable>();
    private final AtomicBoolean ab = new AtomicBoolean(false);

    public void notifies(Runnable r) {
        //Hole den aktuellen Wert
        boolean state = ab.get();

        //Wenn dieser Wert bereits true ist => run()
        if (state) {
            r.run();
            return;
        }

        //Nun fügen wir hier erstmal r hinzu...
        queue.offer(r);
 
        //Nun überprüfen wir hier, ob AtomicBoolean false ist und erneut auf false gesetzt werden kann... CAS
        if (!ab.compareAndSet(false, false)) {
            //Wenn das nicht der Fall ist (Also eine Unterbrechung stattgefunden hat durch einen anderen Thread...)
            if (queue.remove(r)) { //Versuche das Runnable zu entfernen
                r.run(); //Sollte es funktioniert haben => run(), Wenn nicht, dann wurde das Runnable gerade eben noch ausgeführt!
            }
        }
    }

    public void finish() {
        if (ab.compareAndSet(false, true)) { //Hier nur auf true setzen, wenn es noch nie auf true gesetzt wurde! AtomOp
            Runnable r;

            while ((r = queue.poll()) != null) {
                r.run(); //Solange run() bis queue leer ist!
            }
        }
    }
}

Meine Fragen dazu:

A) Ist das hier eine korrekte Impl. um Thread-Safe zu garantieren ?
B) Sind die Bedingungen erfüllt ?
-Jedes Runnable darf/muss maximal einmal ausgeführt werden!
-finish() kann nur von maximal einem Thread auf true gesetzt werden! compareAndSet sollte dies verhindern!
-notifies(Runnable r) sollte eigentlich von beliebig vielen Threads parallel ausgeführt werden dürfen!


Die finale Frage ist aber : Lohnt sich hier der Stress ? Wäre ein simples synchronized besser/schneller ? Wohlgemerkt kann diese Klasse durchaus im großen Stil benutzt werden in meinem Fall! Also mehrere 100 Runnable könnten evt. in gewissen Abständen nacheinander mit notifies() hinzugefügt werden! Immer aus anderen Threads theoretisch...

So, ich hoffe ihr könnt mir da etwas helfen bzw Licht ins Dunkel bringen! Vlt. übertreibe ich ja auch und Synchroniserung ist garnicht so gravierend...

Die Infos, die ich habe, habe ich hierher:

JAX TV: Java-Programmierung im Multicore-Zeitalter

Gruß,

Chris

 

[Sekundentakt]

Hi Chris,

if (!ab.compareAndSet(false, false))

Damit prüfst Du , ob der aktuelle Wert False ist. Wenn er False ist, setzt Du ihn wieder False.
Warum fragst Du nicht "if(!ab.get())?
Damit überprüfst Du den aktuellen Wert und sparst Dir das Set.

//Wenn das nicht der Fall ist (Also eine Unterbrechung stattgefunden hat durch einen anderen Thread...)

Das musst Du jetzt noch mal erklären. Wenn die Variable "ab" schon mal auf true gesetzt worden ist, führst Du das Runnable aus. Du garantierst aber nicht, dass das nur einmal ausgeführt werden kann, weil du ab nicht wieder auf false setzt, wenn es schon mal true war.
Oder versuchst Du in der Abfrage aus Zeile 26 genau darauf zu überprüfen? War das Ding schon einmal true, und wenn ja entferne es aus der Queue, damit es nicht noch mal ausgeführt werden kann???

In finish() checkst Du, ob ab false war und jetzt true gesetzt werden kann. Wenn es true gesetzt worden ist, führst Du die Runnables aus. Hmm,..

Ich glaube, dass man zwischen Zeile 17 und 20 das Runnable mehrmals ausführen kann.
Nämlich dann, wenn der Prozessor nach Zeile 20 den Thread kurzzeitig unterbricht, einen anderen Thread arbeiten lässt und dieser wieder bis zu Zeile 17 gelangt.
In diesem Falle wäre das Runnable nämlich noch immer in der Queue, dann sogar zwei mal. Da es sich dann theoretisch sogar zwei mal in der Queue befinden könnte, würde ein queue.remove(r) auch nur eine Instanz entfernen - d.h. das andere, ggf. gleichwertige Runnable, könnte sich noch mal in der Queue befinden.
Du solltest also checken, ob sich r bereits in der Queue befindet, bevor Du es einfügst.

Und für das Problem zwischen Zeile 17 und 20 würde ich vielleicht ein weiteres Flag einführen, nämlich eines das überprüft, ob dieser Codeblock gerade ausgeführt wird.

Da das Thema sehr komplex ist, sind alle meine Aussagen ohne Gewähr und definitiv diskussionswürdig.

EDIT:

boolean isRunning = atomicIsRunning.compareAndSet(false, true) //wenn es false war, wird es jetzt zum ersten mal ausgeführt, ansonsten können wir das hier überspringen
if(isRunning)
{
       //Wenn dieser Wert bereits true ist => run()
        if (state) {
            r.run();
            return;
        }
}

Macht das Sinn?

 

[Kr0e]

Hallo Sekundentakt,

erstmal vielen Dank, dass du dir am Sonntag die Zeit nimmst, dich mit einem derart schweren Thema auseinander zu setzen!

Also kurz zur Erläuterung: finish() darf maximal einmal ausgeführt werden! Alle weiteren sollen ja direkt bei "notifies" dann ausgeführt werden! Da ich die ConcurrentLinkedQueue nehme, sind auch die Methoden offer/poll/remove alle atomar und ununterbrechbar...
Die ConcurrentLinkedQueue benutzt dafür auch diese CompareAndSwap Funktionalität und kommt ganz ohne Sync aus...
Von daher wollte ich das auch iwie ohne schaffen. Jetzt zu deinen Ideen:

Dass das hier "if (!ab.compareAndSet(false, false))" mit "if(!ab.get())" ausgetauscht werden kann, sehe ich ein! Da hab ich einen Denkfehler gemacht. Danke =)

Aus meiner Sicht, ist es nicht möglich, dass EIN Runnable zweimal in einer List vorkommt, nur wenn man 2x notifies(r) aufruft vlt...
Wenn "state" true ist, wird es niemals mehr false (einmal finish... immer finish) und wenn state == true ist, dann folgt auch ein "return" womit in diesem Fall Zeile 20 garnicht mehr erreicht wird. Aber kann gut sein, dass ich deinen Hinweis grad nicht begriffen habe!

Obwohl ich mich jetzt schon relativ lange mit diesem Thema beschäftige, fällt es mir immer noch sehr schwer, in dieser Abstraktion (Ohne Locks, etc..) produktiven Code zu schreiben. Ich hab das Gefühl, da müsste iwie ein Lock herein!

Ansonsten... Da du vlt. fitter bist als ich... Wie würdest du dieses Problem allgemein angehen ? (Oder auch alle Anderen die das lesen ) Es ist ja eigentlcih ein relativ "simples" Problem gefolgt von einer hochkomplexen Lösung!

Falls du Lust hast und Zeit und ne gute Idee, wäre ich dir sehr dankbar wenn du vlt eine bessere (Und vorallem sichere Lösung) skizzieren oder vlt sogar impl könntest, es geht ja nur um den Knackpunkt wo es eben das Problem gibt!

Nämlich -> Wenn notifies() und finish() parallel aufgerufen werden! In allen and eren Fällen kommt es ja zu garkeinem Problem...
Und nur für diesen Fall müsste man dann ja eine (möglichst "lock-free") Alternative finden...

Dann schonmal ein dickes Dankeschön!

Gruß,

Chris

 

[Sekundentakt]

Hi Chris,

um ehrlich zu sein, habe ich Lock-Free noch nie programmiert, weil ich dieses Feature noch nie benötigt habe. Die Theorie fand ich aber interessant und habe mich damit mal beschäftigt.
Ich glaube also, dass Du mehr Erfahrung darin haben wirst, als ich.

Wenn "state" true ist, wird es niemals mehr false (einmal finish... immer finish) und wenn state == true ist, dann folgt auch ein "return" womit in diesem Fall Zeile 20 garnicht mehr erreicht wird. Aber kann gut sein, dass ich deinen Hinweis grad nicht begriffen habe!

Ich glaube da missverstehen wir uns.

Stell Dir einfach vor, die Prozessoren Deines Systems sind voll ausgelastet. Mehreren Threads erhalten also stückweise Zeit im Prozessor. Angelika Langer erklärt dies als das "alte" denken. Prinzipiell hat sie damit recht - wenn zwei Prozessorkerne existieren, können auch zwei Threads parallel ausgeführt werden. Da auf einem System aber global nie nur zwei Threads laufen, sondern viele (das OS hat ja selbst auch einige Prozesse am laufen), teilen die Prozessorkerne jedem Thread immer noch stückchenweise Zeit zu.
Ein Beispiel: Prozessorkern A lässt deinen aktuellen Thread bis zur Zeile 17 laufen, merkt sich dabei das die Abfrage TRUE ergeben hat. Stell Dir vor jetzt entzieht er dem aktuellen Thread das "Arbeitsrecht" und gibt es an einen anderen Thread, der genau die selbe Methode aufruft.
Diese Thread könnte jetzt theoretisch die gesamte Methode durchlaufen.
Wenn Prozessorkern A dem alten Thread das Arbeitsrecht wieder erteilt, macht er dort weiter, wo er zuvor aufgehört hat - er führt jetzt erst den Inhalt des if-Statements aus.
Ich weiß nicht wie es sich mit dem Rest verhält - aber in diesem Falle wird run zwei mal beim möglicherweise selben Runnable aufgerufen.
Anders ist es aber, wenn Du meinen Lösungsvorschlag nimmst.
Wenn die selbe Situation mit meinem Lösungsvorschlag geschehen würde, käme der zweite Thread gar nicht mehr in das if-Statement rein, weil er zuvor an der isRunning-Condition scheitern wird. Denn die wird jetzt beim isRunning.compareAndSet()-Befehl false zurückgeben.

EDIT:
Ich gehe jetzt aber davon aus, dass man notifies() nur einmal im gesamten Event aufrufen darf.
Falls nicht, musst du atomicIsRunning nach Ausführung der "gefährlichen" Codefragmente natürlich wieder in seinen Eingangs-Wert umschreiben. Also atomicIsRunning.compareAndSet(true, false) aufrufen.
Ansonsten würde meine Methode außerdem dazu führen, dass maximal ein Runnable zur selben Zeit - unabhängig von seinem Typ - mit diesem Event-Objekt ausgeführt werden darf.
Ich weiß nicht, ob das beabsichtigt ist.


EDIT²: Wenn ich mich recht entsinne hat aber auch Frau Langer bei dem von Dir genannten Link erklärt, dass man in bestimmten Situationen nicht um Locks oder Lock-ähnliche Situationen herumkommt, wenn man bestimmte Dinge garantieren will. Könntest Du vielleicht noch mal den Unterschied zwischen dem was bei notifies() bei state = true ausgeführt werden soll, und dem was bei finishes gemacht wird, erklären? state=true wäre ja nur dann der Fall, wenn finishes bereits aufgerufen worden ist, oder?

Removes a single instance of the specified element from this queue, if it is present. More formally, removes an element e such that o.equals(e), if this queue contains one or more such elements. Returns true if this queue contained the specified element (or equivalently, if this queue changed as a result of the call).

Das steht in der javadoc zum Remove-Befehl. Du musst also vorher checken, ob bereits ein gleichwertiges Runnable (via contains()) in der Queue steckt, wenn du umgehen willst, dass ein und das selbe Runnable zwei mal in der Queue stecken könnte.

Bis auf diesen Knackpunkt, sehe ich im Moment keine weiteren Logiklücken.

Wie siehst Du das?

 

[Kr0e]

Also: notifies darf beliebig oft mit beliebig unterschiedlichen (oder auch gleichen!) Runnables ausgeführt werden!
finish() darf nur einmal ausgeführt werden. Wird es nochmal ausgeführt, sollte die Funktion einfach nichts machen!

Ich werde mal deinen Ansatz probieren mit dem Statusflag... Danke =)


EDIT:

Zu deinem EDIT2:

Also: Da finish nur einmal aufgerufen werden kann, würden zwangsläufig alle Runnables die quasi nach einem Aufruf von finish hinzugefügt werden, einfach direkt ignoriert und niemals benachrichtigt, weil eben finish schon fertig is... Und genau diese sollen dann
direkt bei notifies() schon benachrichtigt werden. In dem anderen Fall soll eben das Runnable noch in die Queue gesetzt werden, bis dann iwann finish() kommt...

 

[Sekundentakt]

Nach Deiner momentanen Implementierung wäre nach dem Aufruf von finish() state = true.
Das heißt jedes Runnable welches notifies() übergeben wird, wird auch definitiv sofort ausgeführt, sofern zuvor finish() ausgeführt worden ist.

Anschließend wird es der Queue hinzugefügt - egal ob state = true, oder nicht.
Und wenn ab.get() nicht false ist, wird es wieder aus der Queue entfernt.
Und da das Runnable so oder so immer in die Queue eingefügt worden wäre, kann man es auch wieder aus der Queue entfernen. Da dies wiederum immer true ergeben würde, wird das Runnable also noch mal ausgeführt.

Hier hast Du m.E. einen Logikfehler.

EDIT:
Ich muss nochmal nachfragen:
Du sagst es ist okay, mehrfach das selbe Runnable-Objekt über notifies() auszuführen.
Ist es auch okay, mehrfach das selbe Runnable-Objekt in der Queue zu haben???

Wenn Du dreimal nacheinander notifies aufrufst und jedesmal das selbe Runnable übergeben wird, steht es auch 3x in der Queue, sofern state != true ist.
Das heißt, wenn jemand finishes aufruft, wird 3x das selbe Runnable aus der Queue geholt.
Wenn Du willst, dass das selbe Runnable nur einmal in der Queue vorkommt, dann musst Du vorher noch checken, ob sich ein solches bereits in der Queue befindet.

 

[Kr0e]

Nein, denn wenn state == true ist, folgt ein return und die Methode ist bei Zeile 19 zuende!

 

[Sekundentakt]

Stimmt, das habe ich übersehen.

Bleibt aber dennoch die Frage mit doppelten Einträgen in der Queue vor dem ersten Aufruf von finishes().

 

[ice-breaker]

ehrlich gesagt kann ich deinen Code an fast keiner Stelle nachvollziehen, besonders wird auch nirgends ersichtlich wann oder wie die finish-Methode aufgerufen wird.

Was mir aber sofort auffällt:
Du hast eine Race-Condition beim Ausführen des Runnables wenn state=true, wenn 2 Threads exakt gleichzeitig laufen, dann werden beide ausgeführt, das entspricht nicht wirklich dem Verhalten einer Queue Oder soll es so sein, wenn das Event einmal ausgelöst wurde, dass die Runnables nicht mehr warten müssen?

Eventuell schonmal an Futures gedacht? Oder einfach einen Executor-Service mit nur 1 Thread zu nutzen?

 

[Kr0e]

Hallo Ice-Breaker!

Genau so ist es. Wenn ein Event (Ereignis) eingetreten ist, soll es niemals mehr eintreten (Neues Event erstellen dann halt...)
Alle die nachträglich darauf bauen, davon benachrichtigt zu werden, müssen dann sofort weiterarbeiten, da das Event bereits eingetroffen ist.

PS:

Sry, dass das etwas unübersichtlich ist. Das ist nur ein kleiner Ausschnitt aus der Datei. Aber nur diese beiden Methoden haben halt mit dem Problem zu tun...

 

[Kr0e]

Hallo,

so jetzt habe ich mal eine Idee implementiert, die mir als lockfree Lock vorschwebt...

Hier wird auf Synchronisierung erzichtet sondern einfach solange in einer forschleife gewartet, bis
der Lock frei ist!

Ich habe es mit einer Counter-Impl getestet. Typisches Consumer/Producersystem... 2 Threads....
Der neue Lock hier ist bei mir 3-4 mal schneller! Habe einen Quadcore.

package foxnet.system.util.concurrent.locks;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;

/**
 *
 * @author Christopher Probst
 */
public class AtomicLock implements Lock {

    private final AtomicBoolean busy = new AtomicBoolean(false);

    private void internalLock() {
        for (;;) {
            if (tryLock()) {
                return;
            }
        }
    }

    public void lock() {
        //Acquire lock
        internalLock();
    }

    public void lockInterruptibly() throws InterruptedException {
        throw new UnsupportedOperationException("Not supported yet.");
    }

    public Condition newCondition() {
        throw new UnsupportedOperationException("Not supported yet.");
    }

    public boolean tryLock() {
        //Try to acquire the lock!
        return busy.compareAndSet(false, true);
    }

    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        throw new UnsupportedOperationException("Not supported yet.");
    }

    public void unlock() {
        //We are not busy now
        busy.set(false);
    }
}

 

[Kr0e]

Dieses Problem kann nun als gelöst angesehen werden.

Mein Event sieht nun so aus und nutzt den AtomicLock:

package foxnet.system.util.concurrent.event;

import foxnet.system.util.concurrent.locks.AtomicLock;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.locks.Lock;

public class Event {

    //Die Runnable Eventqueue
    private final ConcurrentLinkedQueue<Runnable> queue 
            = new ConcurrentLinkedQueue<Runnable>();

    //Der neue atomare Lock
    private final Lock atomicLock = new AtomicLock();

    //Event bereits eingetreten ?
    private volatile boolean complete = false;

    public void notifies(Runnable r) {
        boolean invokeNow = false;

        //Zuerst lock sichern
        atomicLock.lock();

        //Wenn wir noch nicht fertig sind,
        //in die Queue aufnehmen...
        if(!complete) {
            queue.add(r);
        } else {
            //Wir sind fertig, aber
            //nicht im Lock aufrufen!
            invokeNow = true;
        }

        //Lock wieder freigeben
        atomicLock.unlock();

        //Direkter call
        if(invokeNow) {
            r.run();
        }
    }

    public void finish() {
        //Zuerst lock sichern
        atomicLock.lock();

        //Jetzt ist das Ereignis fertig
        complete = true;

        //Lock wieder freigeben
        atomicLock.unlock();

        //Tmp
        Runnable run;

        //Alle Runnables auslösen
        while( (run = queue.poll()) != null ) {
            run.run();
        }
    }
}

In meinem Messagingsystem passt dieser AtomicLock hervoragend, da an vielen Ecken nur sehr kurz gewartet werden muss und
von daher die kurze Dauerschleife schneller ist als ein Thread-Sleep durch sync/explicit Locks von Java.

Danke für die Hilfe!