Generator mit virtuellem Thread

[Barista]

Seltsamerweise findet man (fand ich) im Netz kaum Beispiele für Generatoren, die es in anderen Programmiersprachen gibt oder die man zum Beispiel in C# mit yield leicht realisieren kann.

Es geht darum, dass man mehrere Werte (Objekte) in einem Iterator zurück gibt, wobei der Iterator zwischen den next-Aufrufen den aktuellen Status vermerken und wieder aufnehmen muss. Man kann nicht einfach eine for-Schleife hinschreiben. Rekursiv wird es ganz übel (Iterator über Baum-Elemente).

Ich weiß nicht, ob es mit Streams eine bessere Lösung gibt.

Mit den neuen virtuellen Threads sollte dies einfach sein:

package kh.gen_virt_thrd;

import java.util.Iterator;
import java.util.NoSuchElementException;
import java.util.concurrent.LinkedBlockingDeque;

/**
 * Abstract class to generate values
 * with support of virtual thread.
 */
abstract public class GeneratorWthVrtThrd<T>
implements Iterable<T>
{
    /**
     * Method to implement to generate values.
     *
     * @param queue the output queue for the generated values
     */
    abstract public void generate(
            final GeneratorDeque<T> queue );

    @Override
    public final Iterator<T> iterator()
    {
        final GeneratorDeque<T> newQueue = new GeneratorDeque<T>();

        Thread.ofVirtual().start( () -> this.generate( newQueue ) );

        return new Iterator<T>()
        {
            private final GeneratorDeque<T> queue = newQueue;

            @Override
            public boolean hasNext()
            {
                return ! this.queue.isClosed();
            }

            @Override
            public T next()
            {
                if ( ! this.hasNext() )
                {
                    throw new NoSuchElementException();
                }

                return this.queue.take();
            }
        };
    }

    /**
     * Queue class to transfer generated values to consumer.
     */
    public static class GeneratorDeque<T>
    {
        private final LinkedBlockingDeque<T> innerQueue = new LinkedBlockingDeque<T>( 1 );

        private boolean isClosed;

        public void close()
        {
            this.isClosed = true;
        }

        public boolean isClosed()
        {
            return this.isClosed;
        }

        /**
         * Put value in queue.
         * Blocks if queue has no space for specified value.
         *
         * @param e value to put
         */
        public void put(
                final T e )
        {
            if ( this.isClosed )
            {
                throw new IllegalStateException( "queue already closed" );
            }

            try
            {
                this.innerQueue.put( e );
                // bei der Variante mit GeneratorDeque als top level Klasse war dieses sleep notwendig
                //Thread.sleep( 1 );
            }
            catch ( final InterruptedException exc )
            {
                throw new RuntimeException( exc );
            }
        }

        /**
         * Takes value from queue.
         * Blocks if queue has no value.
         *
         * @return value from queue
         */
        public T take()
        {
            if ( this.isClosed )
            {
                throw new IllegalStateException( "queue already closed" );
            }

            try
            {
                return this.innerQueue.take();
            }
            catch ( final InterruptedException exc )
            {
                throw new RuntimeException( exc );
            }
        }
    }

}

Hier der Code in Anwendung:

package kh.gen_virt_thrd;

/**
 * Spike for {@link GeneratorWthVrtThrd}.
 */
public class Integer0to2Generator
extends GeneratorWthVrtThrd<Integer>
{
    /**
     * @param args
     */
    public static void main(final String[] args)
    {
        // use output buffer to avoid blocking main thread for io operations
        final StringBuilder buff = new StringBuilder();

        for ( final Integer i : new Integer0to2Generator() )
        {
            //System.out.println( i );
            buff.append( i );
            buff.append( '\n' );
        }

        System.out.println( buff );
    }

    @Override
    public void generate(
            final GeneratorDeque<Integer> queue )
    {
        for ( int i = 0 ; i < 3 ; i++ )
        {
            //System.out.println( "queue.put: " + i );
            queue.put( i );
            //Thread.sleep( 1 );
        }
        //System.out.println( "queue.close" );
        queue.close();
    }

}

Erst hatte ich die static encapsulated Klasse GeneratorDeque als eigene top level Klasse.

Da hat der Code nur funktioniert, wenn ich das Thread.sleep( 1 ) aufgerufen habe (mit Thread.sleep( 0 ) ) hat es auch nicht funktioniert).

Der main-Thread hat blockiert, Es gab nur die Ausgaben 0, 1 und dann nix mehr, aber das Programm lief noch.

Komischerweise musste ich Thread.sleep( 1 ) nicht mehr aufrufen, als ich die Klasse GeneratorDeque zur eingeschlossenen Klasse gemacht habe.

Ich bin mir jetzt unsicher, ob das Konstrukt nicht doch irgendwann ungewollt blockiert.

Die jetzt auskommentierten System.out.println haben den Code auch funktionieren lassen, klar, bei IO blockiert der virtuelle Thread, gibt also die Kontrolle ab.

Wie kann ich es wirklich richtig machen?

 

[Oneixee5]

Wenn ich das richtig sehe willst du erreichen, dass deine sequentielle Eingabe (put) - zwar in einem anderen Thread - trotzdem wieder sequentiell abgearbeitet wird. Das ganze in einem virtuellen Thread. Nun ist es so, das hier ein Thread eigentlich gar nicht notwendig wäre aber die willst einen non-blocking-input erreichen. Die Ausgabe soll dann im Main-Thread erfolgen.
Virtuelle Threads sind für die kurzzeitige Nutzung mit einer konkreten kompakten Aufgabe gedacht. Es werden direkt Vorteile der CPU genutzt und es entsteht minimaler Overhead. Um sich an diese Konzept zu halten, solltest du bei jedem put statt der Queue jedes Mal einen neuen virtuellen Thread erzeugen. Allerdings ist dann nicht garantiert, dass die Ausgabe in der Reihenfolge der Eingabe erfolgt.

Zur Funktion: LinkedBlockingDeque implementiert bereits Iterable, von daher ist GeneratorDeque eigentlich überflüssig. Für die Funktonalität queue.close(); sehe ich keine Notwendigkeit. Der Ausgabe-Thread ist hier der Main-Thread. Dieser wird also blockiert, wenn keine Eingaben in der Queue vorhanden sind. Das sehe ich problematisch, denn damit blockiert man diesen. Nur deshalb benötigst du close(); Wäre es nicht besser, wenn die Ausgabe in einem eigenem Thread ablaufen würde? Damit wäre der Main-Thread nie blockiert.

Ich denke die Struktur ist hier falsch gewählt. Es sollte natürlich eine BlockingQueue geben aber eingaben sollten von mind. einem Producer in der Queue abgelegt werden und dann von einem oder mehr Consumern der Queue wieder entnommen werden. Das Ende des Programms wird dadurch ermöglicht, dass auf allen noch laufenden Producer-Threads und Consumer-Threads interrupted aufgerufen wird. Dabei ist es praktisch, wenn man für die Erzeugung Threadgroup verwendet, so behält man die Kontrolle über alle eigene laufenden Threads.