java - progressbar - SwingWorker, done() se ejecuta antes de que finalicen las llamadas a process()

swingworker java ejemplo

Question

He estado trabajando con SwingWorker s por un tiempo y terminé con un comportamiento extraño, al menos para mí. Entiendo claramente que, debido a razones de rendimiento, varias invocaciones al método publish() están en una invocación. Tiene mucho sentido para mí y sospecho que SwingWorker mantiene algún tipo de cola para procesar todas las llamadas.

De acuerdo con el tutorial y la API, cuando SwingWorker finaliza su ejecución, doInBackground() termina normalmente o el subproceso de trabajo se cancela desde el exterior, luego se invoca el método done() . Hasta ahora tan bueno.

Pero tengo un ejemplo (similar al que se muestra en los tutoriales) donde se ejecutan las llamadas al método process() después de que se ejecuta el método done() . Dado que ambos métodos se ejecutan en el evento Dispatch Thread , esperaría que done() se ejecutara después de que todas las invocaciones de process() hayan finalizado. En otras palabras:

Esperado:

Writing...
Writing...
Stopped!

Resultado:

Writing...
Stopped!
Writing...

Código de muestra

import java.awt.BorderLayout;
import java.awt.Dimension;
import java.awt.Graphics;
import java.awt.event.ActionEvent;
import java.util.List;
import javax.swing.AbstractAction;
import javax.swing.Action;
import javax.swing.JButton;
import javax.swing.JFrame;
import javax.swing.JPanel;
import javax.swing.JScrollPane;
import javax.swing.JTextArea;
import javax.swing.SwingUtilities;
import javax.swing.SwingWorker;

public class Demo {

    private SwingWorker<Void, String> worker;
    private JTextArea textArea;
    private Action startAction, stopAction;

    private void createAndShowGui() {

        startAction = new AbstractAction("Start writing") {
            @Override
            public void actionPerformed(ActionEvent e) {
                Demo.this.startWriting();
                this.setEnabled(false);
                stopAction.setEnabled(true);
            }
        };

        stopAction = new AbstractAction("Stop writing") {
            @Override
            public void actionPerformed(ActionEvent e) {
                Demo.this.stopWriting();
                this.setEnabled(false);
                startAction.setEnabled(true);
            }
        };

        JPanel buttonsPanel = new JPanel();
        buttonsPanel.add(new JButton(startAction));
        buttonsPanel.add(new JButton(stopAction));

        textArea = new JTextArea(30, 50);
        JScrollPane scrollPane = new JScrollPane(textArea);

        JFrame frame = new JFrame("Test");
        frame.setDefaultCloseOperation(JFrame.DISPOSE_ON_CLOSE);
        frame.add(scrollPane);
        frame.add(buttonsPanel, BorderLayout.SOUTH);
        frame.pack();
        frame.setLocationRelativeTo(null);
        frame.setVisible(true);
    }

    private void startWriting() {
        stopWriting();
        worker = new SwingWorker<Void, String>() {
            @Override
            protected Void doInBackground() throws Exception {
                while(!isCancelled()) {
                    publish("Writing.../n");
                }
                return null;
            }

            @Override
            protected void process(List<String> chunks) {
                String string = chunks.get(chunks.size() - 1);
                textArea.append(string);
            }

            @Override
            protected void done() {
                textArea.append("Stopped!/n");
            }
        };
        worker.execute();
    }

    private void stopWriting() {
        if(worker != null && !worker.isCancelled()) {
            worker.cancel(true);
        }
    }

    public static void main(String[] args) {
        SwingUtilities.invokeLater(new Runnable() {
            @Override
            public void run() {
                new Demo().createAndShowGui();
            }
        });
    }
}

Answer 1

Después de leer la magnífica respuesta de DSquare y concluir de que se necesitarían algunas subclases, se me ocurrió esta idea para cualquiera que necesite asegurarse de que todos los fragmentos publicados se hayan procesado en la EDT antes de continuar.

NB Intenté escribirlo en Java en lugar de Jython (mi idioma de elección y oficialmente el mejor idioma del mundo), pero es un poco complicado porque, por ejemplo, publish es final , así que tendrías que inventar otro método. para llamarlo, y también porque tiene que (bostezar) parametrizar todo con los genéricos en Java.

Este código debe ser comprensible para cualquier persona de Java: solo para ayudar, con self.publication_counter.get() , esto se evalúa como False cuando el resultado es 0.

# this is how you say Worker... is a subclass of SwingWorker in Python/Jython
class WorkerAbleToWaitForPublicationToFinish( javax.swing.SwingWorker ):

    # __init__ is the constructor method in Python/Jython
    def __init__( self ):

        # we just add an "attribute" (here, publication_counter) to the object being created (self) to create a field of the new object
        self.publication_counter = java.util.concurrent.atomic.AtomicInteger()

    def await_processing_of_all_chunks( self ):
        while self.publication_counter.get():
            time.sleep( 0.001 )

    # fully functional override of the Java method     
    def process( self, chunks ):
        for chunk in chunks:
            pass
            # DO SOMETHING WITH EACH CHUNK

        # decrement the counter by the number of chunks received
        # NB do this AFTER dealing with the chunks 
        self.publication_counter.addAndGet( - len( chunks ) )

    # fully functional override of the Java method     
    def publish( self, *chunks ):
        # increment the counter by the number of chunks received
        # NB do this BEFORE publishing the chunks
        self.publication_counter.addAndGet( len( chunks ))
        self.super__publish( chunks )

Así que en tu código de llamada, pones algo como:

    engine.update_xliff_task.get()
    engine.update_xliff_task.await_processing_of_all_chunks()

PD: el uso de una cláusula while como esta (es decir, una técnica de sondeo) no es elegante. Miré las clases disponibles de java.util.concurrent como CountDownLatch y Phaser (ambas con métodos de bloqueo de subprocesos), pero no creo que ninguna de las dos sea adecuada para este propósito ...

luego

Me interesó lo suficiente como para modificar una clase de concurrencia adecuada (escrita en Java, que se encuentra en el sitio de Apache) llamada CounterLatch . Su versión detiene el hilo en await() si se alcanza el valor de un contador AtomicLong . Mi versión aquí te permite hacer eso, o lo contrario: decir "esperar hasta que el contador alcance un cierto valor antes de levantar el pestillo":

NB: uso de AtomicLong para signal y AtomicBoolean para released : porque en el Java original usan la palabra clave volatile . Creo que usar las clases atómicas logrará el mismo propósito.

class CounterLatch():
    def __init__( self, initial = 0, wait_value = 0, lift_on_reached = True ):
        self.count = java.util.concurrent.atomic.AtomicLong( initial )
        self.signal = java.util.concurrent.atomic.AtomicLong( wait_value )

        class Sync( java.util.concurrent.locks.AbstractQueuedSynchronizer ):
            def tryAcquireShared( sync_self, arg ):
                if lift_on_reached:
                    return -1 if (( not self.released.get() ) and self.count.get() != self.signal.get() ) else 1
                else:
                    return -1 if (( not self.released.get() ) and self.count.get() == self.signal.get() ) else 1
            def tryReleaseShared( self, args ):
                return True

        self.sync = Sync()
        self.released = java.util.concurrent.atomic.AtomicBoolean() # initialised at False

    def await( self, *args ):
        if args:
            assert len( args ) == 2
            assert type( args[ 0 ] ) is int
            timeout = args[ 0 ]
            assert type( args[ 1 ] ) is java.util.concurrent.TimeUnit
            unit = args[ 1 ]
            return self.sync.tryAcquireSharedNanos(1, unit.toNanos(timeout))
        else:
            self.sync.acquireSharedInterruptibly( 1 )

    def count_relative( self, n ):
        previous = self.count.addAndGet( n )
        if previous == self.signal.get():
            self.sync.releaseShared( 0 )
        return previous

Así que mi código ahora se ve así:

En el constructor SwingWorker:

self.publication_counter_latch = CounterLatch() 

En SW.publish:

self.publication_counter_latch.count_relative( len( chunks ) )
self.super__publish( chunks )

En el hilo esperando que se detenga el procesamiento de trozos:

worker.publication_counter_latch.await()

Answer 2

RESPUESTA CORTA:

Esto sucede porque publish () no programa directamente el process , establece un temporizador que activará la programación de un bloque process () en la EDT después de DELAY . Entonces, cuando el trabajador se cancela, todavía hay un temporizador esperando para programar un proceso () con los datos de la última publicación. La razón para usar un temporizador es implementar la optimización donde se puede ejecutar un proceso único con los datos combinados de varias publicaciones.

RESPUESTA LARGA:

Veamos cómo publicar () y cancelar interactúan entre sí, para eso, vamos a sumergirnos en algún código fuente.

Primero la parte fácil, cancel(true) :

public final boolean cancel(boolean mayInterruptIfRunning) {
    return future.cancel(mayInterruptIfRunning);
}

Esta cancelación termina llamando al siguiente código:

boolean innerCancel(boolean mayInterruptIfRunning) {
    for (;;) {
        int s = getState();
        if (ranOrCancelled(s))
            return false;
        if (compareAndSetState(s, CANCELLED)) // <-----
            break;
    }
    if (mayInterruptIfRunning) {
        Thread r = runner;
        if (r != null)
            r.interrupt(); // <-----
    }
    releaseShared(0);
    done(); // <-----
    return true;
}

El estado de SwingWorker se establece en CANCELLED , el hilo se interrumpe y se ejecuta done() , sin embargo, esto no es SwingWorker, sino el future hecho (), que se especifica cuando se crea una instancia de la variable en el constructor SwingWorker:

future = new FutureTask<T>(callable) {
    @Override
    protected void done() {
        doneEDT();  // <-----
        setState(StateValue.DONE);
    }
};

Y el código doneEDT() es:

private void doneEDT() {
    Runnable doDone =
        new Runnable() {
            public void run() {
                done(); // <-----
            }
        };
    if (SwingUtilities.isEventDispatchThread()) {
        doDone.run(); // <-----
    } else {
        doSubmit.add(doDone);
    }
}

Que llama a SwingWorkers''s done() directamente si estamos en la EDT, que es nuestro caso. En este punto, el SwingWorker debe detenerse, no se debe llamar a publish() , esto es bastante fácil de demostrar con la siguiente modificación:

while(!isCancelled()) {
    textArea.append("Calling publish/n");
    publish("Writing.../n");
}

Sin embargo, seguimos recibiendo un mensaje de "Escritura ..." del proceso (). Así que veamos cómo se llama el proceso (). El código fuente para publish(...) es

protected final void publish(V... chunks) {
    synchronized (this) {
        if (doProcess == null) {
            doProcess = new AccumulativeRunnable<V>() {
                @Override
                public void run(List<V> args) {
                    process(args); // <-----
                }
                @Override
                protected void submit() {
                    doSubmit.add(this); // <-----
                }
            };
        }
    }
    doProcess.add(chunks);  // <-----
}

Vemos que run() de Runnable doProcess es quien termina el process(args) llamada process(args) , pero este código solo llama a doProcess.add(chunks) no a doProcess.run() y hay un doSubmit alrededor también. Veamos doProcess.add(chunks) .

public final synchronized void add(T... args) {
    boolean isSubmitted = true;
    if (arguments == null) {
        isSubmitted = false;
        arguments = new ArrayList<T>();
    }
    Collections.addAll(arguments, args); // <-----
    if (!isSubmitted) { //This is what will make that for multiple publishes only one process is executed
        submit(); // <-----
    }
}

Entonces, lo que publish() hace en realidad es agregar los fragmentos a algunos arguments internos de ArrayList y llamar a submit() . Acabamos de ver que submit simplemente llama a doSubmit.add(this) , que es el mismo método para add , ya que doProcess y doSubmit extienden AccumulativeRunnable<V> , sin embargo, esta vez, V es Runnable lugar de String como en doProcess . Así que un trozo es el ejecutable que llama process(args) . Sin embargo, la llamada a submit() es un método completamente diferente definido en la clase de doSubmit :

private static class DoSubmitAccumulativeRunnable
     extends AccumulativeRunnable<Runnable> implements ActionListener {
    private final static int DELAY = (int) (1000 / 30);
    @Override
    protected void run(List<Runnable> args) {
        for (Runnable runnable : args) {
            runnable.run();
        }
    }
    @Override
    protected void submit() {
        Timer timer = new Timer(DELAY, this); // <-----
        timer.setRepeats(false);
        timer.start();
    }
    public void actionPerformed(ActionEvent event) {
        run(); // <-----
    }
}

Crea un temporizador que dispara el código actionPerformed una vez después de DELAY milisegundos. Una vez que se activa el evento, el código se pondrá en cola en la EDT que llamará a un run() interno run() que termina llamando al run(flush()) de doProcess y ejecutando el process(chunk) , donde chunk es el dato vaciado de los arguments Lista de arreglo. Me salté algunos detalles, la cadena de llamadas "correr" es así:

• doSubmit.run ()
• doSubmit.run (flush ()) // En realidad, un bucle de runnables, pero solo tendrá uno (*)
• doProcess.run ()
• doProcess.run (flush ())
• proceso (trozo)

(*) El booleano isSubmited y flush() (que restablece este booleano) hace que las llamadas adicionales para publicar no agreguen ejecutables doProcess para ser llamados en doSubmit.run (flush ()) sin embargo, sus datos no se ignoran. De este modo, se ejecuta un proceso único para cualquier cantidad de publicaciones llamadas durante la vida de un temporizador.

En general, lo que publish("Writing...") hace la programación de la llamada a process(chunk) en la EDT después de un RETARDO. Esto explica por qué incluso después de que cancelamos el subproceso y ya no se realizan más publicaciones, aún aparece un proceso de ejecución, porque en el momento en que cancelamos al trabajador, hay (con alta probabilidad) un temporizador que programará un process() después de haber done() ya está programado

¿Por qué se utiliza este temporizador en lugar de solo programar el proceso () en la EDT con un invokeLater(doProcess) ? Para implementar la optimización de rendimiento explicada en la docs :

Debido a que el método de proceso se invoca de forma asíncrona en el tema de publicación de eventos, se pueden producir múltiples invocaciones al método de publicación antes de que se ejecute el método de proceso. Para fines de rendimiento, todas estas invocaciones se unen en una invocación con argumentos concatenados. Por ejemplo:

 publish("1");
 publish("2", "3");
 publish("4", "5", "6");

might result in:
 process("1", "2", "3", "4", "5", "6")

Ahora sabemos que esto funciona porque todas las publicaciones que ocurren dentro de un intervalo de RETARDO están agregando sus arguments en esa variable interna que vimos arguments y el process(chunk) se ejecutará con todos esos datos de una sola vez.

¿ESTE ES UN ERROR? ¿SOLUCIÓN DE TRABAJO?

Es difícil saber si se trata de un error o no, podría tener sentido procesar los datos que ha publicado el subproceso en segundo plano, ya que el trabajo ya está hecho y es posible que esté interesado en actualizar la GUI con tanta información como pueda. (si eso es lo que hace process() , por ejemplo). Y entonces puede que no tenga sentido si se done() requiere tener todos los datos procesados ​​y / o una llamada a procesar () después de hecho () crea inconsistencias de datos / GUI.

Hay una solución obvia si no desea que se ejecute ningún proceso nuevo () después de terminar (), ¡simplemente verifique si el trabajador está cancelado en el método de process también!

@Override
protected void process(List<String> chunks) {
    if (isCancelled()) return;
    String string = chunks.get(chunks.size() - 1);
    textArea.append(string);
}

Es más difícil hacer que done () se ejecute después de ese último proceso (), por ejemplo, done podría usar también un temporizador que programará el trabajo hecho () después de> DELAY. Aunque no puedo pensar que este sea un caso común, ya que si cancela, no debería ser importante perder un proceso más () cuando sabemos que de hecho estamos cancelando la ejecución de todos los futuros.