Generator mit virtuellem Thread

[Barista]

Guck mal

Continuations: The magic behind virtual threads in Java by Balkrishna Rawool

 

[Barista]

Virtuelle Threads sind auch Threads.

Die Bezeichnung ist auch eine Frage der Definition.

Die normalen Java-Threads haben ihre Wiederspiegelung im Betriebssystem.

Sie haben eigene Stacks.

Sie sind eine Art leichtgewichtige Prozesse (Name im Umfeld Betriebssystem).

Die Green-Threads von früher haben auch Thread im Namen, waren aber keine Threads im Betriebssystem.

Das schwergewichtige bei Threads kommt davon, dass es Betriebssystem-Threads sind.

 

[Barista]

Die zweite Problematik aus meiner Sicht ist: Kenne Dein System!

Aha, Du meinst Java, JDK, Libs und Frameworks.

Da gibt es doch was

GitHub - HeinerKuecker/Coroutine-in-pure-Java: Coroutine as iterator or procedure with yield return only with pure Java

Coroutine as iterator or procedure with yield return only with pure Java - HeinerKuecker/Coroutine-in-pure-Java

github.com

Ups, kommt mir bekannt vor, stammt von mir.

 

[Barista]

Die normalen Java-Threads haben ihre Wiederspiegelung im Betriebssystem.

Platform thread ist der richtige Name.

 

[Barista]

Guck mal


Continuations: The magic behind virtual threads in Java by Balkrishna Rawool

GitHub - HeinerKuecker/Coroutine-in-pure-Java: Coroutine as iterator or procedure with yield return only with pure Java

Coroutine as iterator or procedure with yield return only with pure Java - HeinerKuecker/Coroutine-in-pure-Java

github.com

Ich bin überrascht, ab Minute 18 spricht der Vortragende über Generators.

Ich habe das Video vorher nicht gesehen, gucke ich gerade an.

 

[KonradN]

Die Bezeichnung ist auch eine Frage der Definition.

Wie wäre es, wenn wir im Java Bereich einfach die Begriffe verwenden würden, die von den Verantwortlichen gewählt wurden in den Spezifikationen und in der JavaDoc?

Platform thread ist der richtige Name.

Du bist echt lustig - Was habe ich den geschrieben? Schauen wir einfach einmal nach:

Eine Instanz von Thread kann also sowohl ein "platform thread" oder ein "virtual thread" sein.

Aber super, dass Du meinen Aussagen nun doch folgen kannst. Oder willst Du immer widersprechen bezüglich virtuelle Threads sind, ebenso wie platform Threads eben Threads (Instanzen von java.lang.Thread)? Toll, dass wir nun doch langsam auf einen Stand kommen, etwas mühsam, aber heya: Es geht langsam vorran.

 

[KonradN]

Ich bin überrascht, ab Minute 18 spricht der Vortragende über Generators.

Super, da hast Du dann ja jetzt über Continuations auch einen Weg, da Dir ja die Wege über Streams, Iteratoren und Supplier nicht gefallen. Aber ist doch super, dass dies als Möglichkeiten auch alle genannt wurden im Video.

Ich sehe das als Demo, wie man Continuations nutzen kann - aber ich sehe da bishern keinen wirklichen Sinn. Das schien mir deutlich mehr Code zu sein, als bei Stream mit Supplier oder so notwendig wäre. Aber je nach Anforderungen mag es evtl. auch Sinn machen, etwas in der Art aufzubauen.

 

[mihe7]

Wo ist denn da das Problem? Das mit dem ProduceEvenNumbers ist dann ein einfacher Supplier<Integer>. Das Limit upTo wandert aber in den Stream (so man Streams verwenden möchte) was auch Sinn macht, denn der Generator ist ein generisches erzeugen der geraden Zahlen und das Limitieren ist eine Frage der Nutzung.

Das Problem ist, dass der Iterator Anweisungen ausführt, nachdem der Wert abgerufen wurde.

 

[KonradN]

Ok, so langsam habe ich dann jetzt die Problematik so langsam verstanden. Auch das Generator<T> Beispiel aus dem Video hat da etwas beigetragen.

Ich schreibe da dann einmal für mich hier einen kleinen Abschluss, bei dem ich diverse Dinge einmal aufzeige, da ich vermute, dass diese nicht richtig verstanden oder nicht richtig aufgegriffen wurden.

Genereller Ansatz eines Generators wie er aus meiner Sicht sein sollte. Hintergrund ist, dass Code ja auch testbar sein muss und so. Wenn man es also theoretisch darstellen sollte, dann ist ein Generator ein Supplier von Werten, wobei gilt, dass der Supplier aus einem Status Z dann bei Aufruf einen Wert w erstellt und einen Status Z'. Das sind dann einfache Aufrufe, man kann Statusübergänge und Werte sehr gut testen. Und man hat damit ein funktionales Interface bedient, das sehr universell nutzbar ist. Man kann damit also sehr einfach zu Streams, Iteratoren, ... kommen. Steam.generate wäre da ein Beispiel.

Dann gibt es tatsächlich Situationen, bei denen man einen komplexen Generator hat, bei dem dies so nicht abbildbar ist. Mir fällt da kein gutes Beispiel ein, also sehe ich einfach ein Programm von mir als Generator von Logmeldungen. Bei dem Beispiel wird deutlich: Da kann man diese Sichtweise nicht wirklich so sehen. Was hier vom Prinzip die Lösung ist, ist dann die Umkehr. Ich habe keinen Supplier von Logmeldungen sondern statt dessen habe ich einen Consumer von Logmeldungen. Damit kann ich beliebige Logmeldungen in beliebigen Code schreiben. Das setzt aber nun voraus, dass ich einen Consumer haben kann.

Wenn ich nun den Fall betrachte, dass beides so nicht zutreffend ist, dann kommt tatsächlich dieser Ansatz mit einem Thread ins Spiel. Ein interessanter Ansatz ist dabei der Generator aus dem Video, aber hier werden JDK Internas verwendet. Das ist aus meiner Sicht nicht gut und ich würde das eher ablehnen. Ich verweise einfach einmal auf JEP 403 statt selbst eine Begründung zu schreiben.

Also bleibt dann der Weg über einen (virtuellen) Thread. Den können wir dann direkt einmal bauen mit dem Rahmen, den es schon gab. Der Generator hat den Consumer Parameter um darüber den Code auszuführen und der Möglichkeit, auf source ein yield aufzurufen. So ein erster Ansatz könnte dann z.B. sein:

package de.kneitzel;

import java.util.Iterator;
import java.util.function.Consumer;

public class Generator<T> implements Iterator<T> {
    private final Source source;
    private final Object lock = new Object();
    private valitile boolean completed = false;

    public Generator(Consumer<Source> consumer) {
        source = new Source();
        Thread.ofVirtual().start(() -> {
            consumer.accept(source);
            synchronized (lock) {
                completed = true;
                lock.notifyAll();
            }
        });
    }

    public boolean hasNext() {
        synchronized (lock) {
            return !completed || source.value != null;
        }
    }

    public T next() {
        synchronized (lock) {
            while (source.value == null && !completed) {
                try {
                    lock.wait();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            T result = source.value;
            source.value = null;
            lock.notify();
            return result;
        }
    }
   
    public class Source {
        private T value;

        public void yield(T t) {
            synchronized (lock) {
                while (value != null) {
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
                value = t;
                lock.notify();
            }
        }
    }

    public static void main(String[] args) throws Exception {
        var generator = new Generator<String>(source -> {
            source.yield("A");
            source.yield("B");
            source.yield("C");
        });

        while (generator.hasNext()) {
            System.out.println(generator.next());
        }
    }
}

Man hat also hier nun einen (virtuellen) Thread.


Nehmen wir nun noch eine Problematik hinzu: Es wurde überlegt, wie man denn so einen Generator aufräumen könnte (Szenario a.la. Client sendet Requests an den Server, der Server hält also für den Client den Generator vor aber der Client meldet sich nicht mehr). Die Verwaltung einer Session ist jetzt mal außen vor - ich konzentriere mich rein auf den Generator. Mein Hinweis war hier: AutoClosable. Der Generator bekommt also noch eine close Methode. Und diese muss dann eigentlich nur auf dem Thread ein interrupt aufrufen. Also das wird dann zu etwas wie:

package de.kneitzel;

import java.util.Iterator;
import java.util.function.Consumer;

public class Generator<T> implements AutoCloseable, Iterator<T> {
    private final Source source;
    private final Thread thread;
    private final Object lock = new Object();
    private volatile boolean completed = false;

    public Generator(Consumer<Source> consumer) {
        source = new Source();
        thread = Thread.ofVirtual().start(() -> {
            consumer.accept(source);
            synchronized (lock) {
                completed = true;
                lock.notifyAll();
            }
        });
    }

    public boolean hasNext() {
        synchronized (lock) {
            return !completed || source.value != null;
        }
    }

    public T next() {
        synchronized (lock) {
            while (source.value == null && !completed) {
                try {
                    lock.wait();
                } catch (InterruptedException e) {
                    throw new RuntimeException(e);
                }
            }
            T result = source.value;
            source.value = null;
            lock.notify();
            return result;
        }
    }

    @Override
    public void close() throws Exception {
        if (thread != null) {
            thread.interrupt();
        }
    }

    public class Source {
        private T value;

        public void yiald(T t) {
            synchronized (lock) {
                while (value != null) {
                    try {
                        lock.wait();
                    } catch (InterruptedException e) {
                        throw new RuntimeException(e);
                    }
                }
                value = t;
                lock.notify();
            }
        }
    }

    public static void main(String[] args) throws Exception {
        var generator = new Generator<String>(source -> {
            source.yiald("A");
            source.yiald("B");
            source.yiald("C");
        });

        if (generator.hasNext()) {
            System.out.println(generator.next());
        }
        generator.close();
    }
}

Das mit dem close() braucht evtl. noch etwas mehr Arbeit - evtl. braucht es ein synchronized Block und so. Aber das Prinzip wird erst einmal deutlich.

Nun gab es aber noch einen weiteren Punkt, der erwähnt wurde: Man kann eine vorhandene Java Framework Klasse nehmen. Java kommt mit vielen guten Klassen und da gibt es auch bereits eine Klasse, die eine Queue mit fester Größe darstellt und deren Aufrufe blocken, wenn in eine volle Queue etwas eingestellt werden soll oder wenn aus einer leeren Queue etwas entnommen werden soll. Darauf kann man auch einfach aufbauen, was dann einen deutlich vereinfachten Code bringt:

package de.kneitzel;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.function.Consumer;

public class Generator<T> {
    private final BlockingQueue<T> queue = new ArrayBlockingQueue<>(1);
    private volatile boolean completed = false;

    public Generator(Consumer<Source> consumer) {
        Thread.ofVirtual().start(() -> {
            try {
                consumer.accept(new Source());
            } finally {
                completed = true;
            }
        });
    }

    public boolean hasNext() {
        return !completed || !queue.isEmpty();
    }

    public T next() {
        try {
            T result = queue.take();
            if (result == null) {
                completed = true;
            }
            return result;
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    public class Source {
        public void yield(T t) {
            try {
                queue.put(t);
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }
    }

    public static void main(String[] args) {
        var generator = new Generator<String>(source -> {
            source.yield("A");
            source.yield("B");
            source.yield("C");
        });

        while (generator.hasNext()) {
            System.out.println(generator.next());
        }
    }
}

Das ist jetzt ohne das AutoClosable und mit fester Queue von 1. Aber wenn man mehr Nebenläufigkeit haben will, dann kann man hier auch mehrere Elemente in der Queue erlauben.

Das einfach einmal um in Form von konkretem Code aufzuzeigen, was mit den diversen Hinweisen gemeint war.

Edit: Die Variable completed wird von mehreren Threads benutzt und muss daher volatile sein.

 

[Barista]

Ok, so langsam habe ich dann jetzt die Problematik so langsam verstanden.

Sehr gut.

Ich bitte um Entschuldigung, wenn ich mich manchmal im Ton vergriffen haben sollte.

Zur Anmerkung: synchronized sollte bei virtuellen Threads nicht verwendet werden, weil dies zum sogenannten Pinning, das Blockieren des Carrier-Threads, führt.
Statt dessen sollte ReentrantLock verwendet werden.

 

[Barista]

Dann gibt es tatsächlich Situationen, bei denen man einen komplexen Generator hat, bei dem dies so nicht abbildbar ist. Mir fällt da kein gutes Beispiel ein

Ein Beispiel wäre das durchlaufen eines binären Baumes in sortierte Reihenfolge.

Da muss man ohne yield einen Stack als Status im Iterator/Generator halten.

Es gibt zwar TreeSet im JDK, aber manchmal benötigt man explizit einen eigenen Baum.

Wahrscheinlich gibt es auch mathematische Anwendungen die über Fibonacci, Fakultät, Primzahlen usw. hinausgehen
(ich hatte mal einen Chef, der sagte, wenn jemand usw, schreibt, heißt das, da kommt nichts mehr, stimmt in diesem Falle).

 

[KonradN]

Ein Punkt ist mir noch zugetragen worden, den ich auf die Schnelle beim Zusammenbasteln nicht beachtet habe: Wenn man eine Variable hat, auf die man von mehreren Threads aus zugreifen will, so sollte diese volatile sein:
private volatile boolean completed = false;

Danke @mihe7 für diesen wichtigen Hinweis. Den Code in meinem Post editiere ich dann noch nachträglich.

 

[KonradN]

Dann noch der Hinweis bezüglich dem synchronized - das führt dann z.B. zu einem Code wie:

package de.kneitzel;

import java.util.Iterator;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Consumer;

public class Generator<T> implements AutoCloseable, Iterator<T> {
    private final Source source;
    private final Thread thread;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition valueAvailable = lock.newCondition();
    private final Condition valueConsumed = lock.newCondition();
    private volatile boolean completed = false;

    public Generator(Consumer<Source> consumer) {
        source = new Source();
        thread = Thread.ofVirtual().start(() -> {
            consumer.accept(source);
            lock.lock();
            try {
                completed = true;
                valueAvailable.signalAll();
            } finally {
                lock.unlock();
            }
        });
    }

    public boolean hasNext() {
        lock.lock();
        try {
            while (!completed && source.value == null) {
                valueAvailable.await();
            }
            return source.value != null;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    public T next() {
        lock.lock();
        try {
            while (source.value == null && !completed) {
                valueAvailable.await();
            }
            T result = source.value;
            source.value = null;
            valueConsumed.signal();
            return result;
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException(e);
        } finally {
            lock.unlock();
        }
    }

    @Override
    public void close() {
        if (thread != null) {
            thread.interrupt();
        }
    }

    public class Source {
        private T value;

        public void yiald(T t) {
            lock.lock();
            try {
                while (value != null) {
                    valueConsumed.await();
                }
                value = t;
                valueAvailable.signal();
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                throw new RuntimeException(e);
            } finally {
                lock.unlock();
            }
        }
    }

    public static void main(String[] args) throws Exception {
        try (var generator = new Generator<String>(source -> {
            source.yiald("A");
            source.yiald("B");
            source.yiald("C");
        })) {

            while (generator.hasNext()) {
                System.out.println(generator.next());
            }
        }
    }
}