Lopende Threads annuleren (Java)

[redesign]

Vraagje...

Wanneer je gebruik maakt van ExecutorService en je wilt dat alle threads die deze executor service aanstuurt worden geannuleerd, dus niet alleen de Runnable tasks die nog in de queue staan die nog niet aan een thread zijn aangewezen, maar ook de Runnables die wel al draaien in een aangewezen Thread, wat moet je dan doen om de Threads "netjes" de nek om te draaien?
De bedoeling is dat wanneer ik mijn Android app afsluit (shutdown) ik ook netjes alle threads opruim en dus ook de threads die hardware aansturen netjes laat communiceren naar de externe hardware dat de verbinding wordt verbroken. Volgens mij heet dat in vakjargon een "graceful disconnect". Het gaat mij niet om de "graceful disconnect" zelf, maar om de threads die opgeruimd/gestopt moeten worden in Java.
Dus als ik bijvoorbeeld een "Thread.stopErMaarMee()" commando krijg, dan weet ik in de thread wel dat hoe ik binnen de thread aangeef aan de hardware "ik stop ermee, toedeledokie!", maar niet hoe ik dan deze thread laat stoppen zodat deze kan worden opgeruimd.

De bedoeling is dat ik ALLE threads die in de ExecutorService draaien met één commando aangeef om te stoppen/annuleren en dat dan binnen deze threads ook weer aan subthreads wordt aangegeven om er mee te stoppen.

Ik heb een voorbeeldtje gemaakt, dat helaas niet doet wat ik verwacht, namelijk doorgaan ipv stoppen!

Het voorbeeld in een notendop: mainthread en executor thread, main thread geeft opdracht om een counter van 10 tot 20 te laten tellen. executor thread wacht tussen elke "tel" 1 seconde. Main thread wacht tussen starten en stoppen 4 seconden. Dus teller zou maar tot 10+4 = 14 moeten tellen lijkt mij.
Maar na een interrupt gaat ie gewoon verder??? Huh?

Ik vergeet iets denk ik.... maar wat?

Hier de code:

Java:

package com.example.threads;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Counter {
    static int poolSize = 4; // 4 cores
    ExecutorService pool;

    public Counter() {
        pool = Executors.newFixedThreadPool(poolSize);
    }

    public void startCounter() {
        // uitgevoerd in een nieuwe thread
        pool.execute(new CountTask(10, 20));
    }

    public void stopCounter() throws InterruptedException {
        // laat de uitvoerende thread eerder stoppen
        pool.shutdownNow();
    }
    
    class CountTask implements Runnable {
        private final int startCounter;
        private final int finishCounter;

        CountTask(int startCounter, int finishCounter) {
            this.startCounter = startCounter;
            this.finishCounter = finishCounter;
        }

        public void run() {
            // print counter to console
            for (int counter = this.startCounter; counter <= this.finishCounter; counter++) {
                System.out.println("counter is: " + counter);
                try {
                     // counter-thread 1 sec laten wachten
                    Thread.sleep(1000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    }

    public static void main(String[] args) throws InterruptedException {
        Counter counter = new Counter();
        counter.startCounter(); // uitgevoerd op de main-thread
        Thread.sleep(4000); // main-thread 4 sec laten wachten
        // stop de counter thread vanaf de main-thread.
        counter.stopCounter();
    }
}

Gaarne jullie hulp.

[Hydra]

De 'nette' manier om threads om zeep te helpen is om in die thread bijv een boolean 'alive' op false te zetten. Die boolean check je dan iedere cycle in je run loop. Een basale run method ziet er dan zo uit:

code:

Public void run(){
   While(alive){
       //dostuff
   }
}

En dan ergens een setter om die boolean te zetten.

(Brakke code want telefoon)

[EddoH]

Je interruptedException catch block staat binnen je for-loop. Deze dien je af te vangen buiten je for loop, ofwel in de catch uit je loop te breaken, anders gaat deze loop gewoon door.

Merk op dat om netjes te stoppen je ook moet 'joinen' op al je threads. Ik weet niet of shutdownNow dit doet.

Lees ook dit: http://www.ibm.com/developerworks/java/library/j-jtp05236/ even door.

[ Voor 19% gewijzigd door EddoH op 17-10-2013 15:46 ]

[redesign]

Hydra schreef op donderdag 17 oktober 2013 @ 15:24:
De 'nette' manier om threads om zeep te helpen is om in die thread bijv een boolean 'alive' op false te zetten. Die boolean check je dan iedere cycle in je run loop. Een basale run method ziet er dan zo uit:

code:

Public void run(){
   While(alive){
       //dostuff
   }
}

En dan ergens een setter om die boolean te zetten.

(Brakke code want telefoon)

Dat is alleen de thread stoppen wanneer deze weer de conditie van de while opnieuw evalueert, toch? Ik bedoel meer om de thread ook te stoppen wanneer deze juist bezig is met de code binnen de run() methode, dus zeg maar als deze berichten onderling met een ander apparaat aan het uitwisselen is, dan blijft deze de run() methode uitvoeren.
Het is idd netjes om de while lus te doorbreken met een false, maar soms duurt het uitwisselen van berichten te lang, en dan wil ik toch graag dat de thread om zeep geholpen wordt en netjes opgeruimd qua geclaimd geheugen.

Is dat ook mogelijk?

[EddoH]

Je kunt niet zomaar midden in je executiepad je thread onderbreken nee. Alleen middels InterruptedException bij blocking methods, bijvoorbeeld sleep en wait, of bij het opnieuw evalueren van een boolean zoals Hydra aangeeft.

Again, Google is your friend:
http://docs.oracle.com/ja...PrimitiveDeprecation.html

[redesign]

EddoH schreef op donderdag 17 oktober 2013 @ 15:42:
Je interruptedException catch block staat binnen je for-loop. Deze dien je af te vangen buiten je for loop, ofwel in de catch uit je loop te breaken, anders gaat deze loop gewoon door.

Merk op dat om netjes te stoppen je ook moet 'joinen' op al je threads. Ik weet niet of shutdownNow dit doet.

Lees ook dit: http://www.ibm.com/developerworks/java/library/j-jtp05236/ even door.

Artikel ga ik zo lezen. Eerst even dit:
1. ik throw nu werkelijk waar overal een InterruptedException, puur omdat ik niet weet wat ik er mee moet doen in welke gevallen. Dat krijg je met newbies
Dus als ik jou goed begrijp moet de for lus tussen de try-catch en niet voor de try-catch. Dat snap ik nog, maar dan die InterruptedException, waarvoor wordt die gebruikt? Alleen maar om "in te breken" op de thread wanneer deze zogenaamd bezig is (sleep is toch ook bezig zijn? bezig zijn met wachten). Moet ik dan juist iets in de catch clausule aanroepen die er voor zorgt dat de thread stopt, bijvoorbeeld een boolean alive=false; zetten wanneer ik de hele zooi wrap in een while(alive) block?

2. op welke plekken moet ik NIET een InterruptedException throwen maar juist afvangen, alleen binnen de run() methode? Ik heb nu een InterruptedException op drie plekken: stopCounter(), run() en main().
Mijn gevoel zegt: dat klopt niet

Ik moet nog veel leren

[Janoz]

Het punt is dat je zelf in je runnable daadwerkelijk het stoppen moet implementeren. Dat kan op de manier van Hydra, maar als ej de javadoc doorleest zul je zien dat de shutdownNow methode op alle lopende threads de interupt aanroept. Wat er effectief nu in je code gebeurt is dat de thread.sleep(1000) onderbroken wordt. De exception 'eet je op' en je thread gaat gewoon verder (Als je je proef programma af laat breken op 3.5 seconden zul je zien dat hij in dat geval maar een halve seconde wacht en 13 en 14 dus wat sneller op elkaar volgen)

Java kan niet bepalen wat een goed punt is om de executie te onderbreken dus dit moet je zelf implementeren. Je kunt echter wel redeljik simpel ontdekken dat je onderbroken wordt. Je kunt immers gewoon de isInterupted() aanroepen. Als dat het geval is breek je gewoon de for lus af.

--edit--
Bij nader inzien. In de javadoc staat dat bij het gooien van de interuptedException de state gecleared wordt. Even goed opletten dus.

--edit^2--
Wat hij hieronder zegt dus

[ Voor 10% gewijzigd door Janoz op 17-10-2013 16:19 ]

[EddoH]

Janoz schreef op donderdag 17 oktober 2013 @ 16:09:

Java kan niet bepalen wat een goed punt is om de executie te onderbreken dus dit moet je zelf implementeren. Je kunt echter wel redeljik simpel ontdekken dat je onderbroken wordt. Je kunt immers gewoon de isInterupted() aanroepen. Als dat het geval is breek je gewoon de for lus af.

Pas wel op met isInterrupted. Als er een InterruptedException gethrowed is de interrupted flag gecleared en returned isInterrupted dus gewoon false.

[redesign]

Ik heb het voorbeeld aangepast, volgens mij doe ik het nu goed zo.

Uitkomst is:
'counter is: 10
counter is: 11
counter is: 12
counter is: 13
Klaar.
CountTask is onderbroken!

Ik had verwacht dat "Klaar." en "CountTask is onderbroken!" qua volgorde andersom zou komen. Ik onderbreek vanuit de main thread de CountTask en die wordt onderbroken dus komt in de catch terecht, dus die regel eerst, daarna kom ik weer terug in de main, dus dan pas "klaar" regel daarna. Tot mijn verbazing niet dus.

De nieuwe code:

Java:

package com.example.threads;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Counter {
    static int poolSize = 4; // 4 cores
    ExecutorService pool;

    public Counter() {
        pool = Executors.newFixedThreadPool(poolSize);
    }

    public void startCounter() {
        // uitgevoerd in een nieuwe thread
        pool.execute(new CountTask(10, 20));
    }

    public void stopCounter() {
        // laat de uitvoerende thread eerder stoppen
        pool.shutdownNow();
    }

    class CountTask implements Runnable {
        private boolean cancelled = false;
        private final int startCounter;
        private final int finishCounter;

        CountTask(int startCounter, int finishCounter) {
            this.startCounter = startCounter;
            this.finishCounter = finishCounter;
        }

        public void run() {
            // print counter to console
            while (!cancelled) {
                try {
                    for (int counter = this.startCounter; counter <= this.finishCounter; counter++) {
                        System.out.println("counter is: " + counter);
                        // counter-thread 1 sec laten wachten
                        Thread.sleep(1000);
                    }
                } catch (InterruptedException e) {
                    cancelled = true;
                    System.out.println("CountTask is onderbroken!");
                }
            }
        }
    }

    public static void main(String[] args) {
        Counter counter = new Counter();
        counter.startCounter(); // uitgevoerd op de main-thread
        try {
            // main-thread 4 sec laten wachten
            Thread.sleep(4000);
        } catch (InterruptedException e) {
            // gewoon doorlopen mensen, NIKS aan het handje :)
        }
        // stop de counter thread.
        counter.stopCounter();
        System.out.println("Klaar.");
    }
}

[EddoH]

redesign schreef op donderdag 17 oktober 2013 @ 16:26:
Ik onderbreek vanuit de main thread de CountTask en die wordt onderbroken dus komt in de catch terecht, dus die regel eerst, daarna kom ik weer terug in de main, dus dan pas "klaar" regel daarna. Tot mijn verbazing niet dus.

De interruptedException wordt in je thread uitgevoerd, niet in je main thread. Om netjes af te sluiten moet je in je main thread wachten tot je threads klaar zijn:

EddoH schreef op donderdag 17 oktober 2013 @ 15:42:
Merk op dat om netjes te stoppen je ook moet 'joinen' op al je threads. Ik weet niet of shutdownNow dit doet.

[redesign]

Hoe join ik op al mijn threads precies? Doe ik dat vanuit de main thread of in de stopCounter method? Graag een code voorbeeld.

[EddoH]

Joinen doe je altijd vanuit de thread die moet wachten (een join blockt over het algemeen totdat de te joinen thread klaar is). Ik zie net dat je geen join hebt bij de ExecutorService maar wel een awaitTermination().

Je kan vast zelf wel uitvinden hoe je die moet gebruiken.

edit: nou vooruit, omdat ik zo aardig ben kopieer en plak ik uit de Java docs:

Java:

//The following method shuts down an ExecutorService in two phases, 
//first by calling shutdown to reject incoming tasks, and then calling shutdownNow, if necessary, 
//to cancel any lingering tasks:
 void shutdownAndAwaitTermination(ExecutorService pool) {
   pool.shutdown(); // Disable new tasks from being submitted
   try {
     // Wait a while for existing tasks to terminate
     if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
       pool.shutdownNow(); // Cancel currently executing tasks
       // Wait a while for tasks to respond to being cancelled
       if (!pool.awaitTermination(60, TimeUnit.SECONDS))
           System.err.println("Pool did not terminate");
     }
   } catch (InterruptedException ie) {
     // (Re-)Cancel if current thread also interrupted
     pool.shutdownNow();
     // Preserve interrupt status
     Thread.currentThread().interrupt();
   }
 }

[ Voor 64% gewijzigd door EddoH op 17-10-2013 16:59 ]

[redesign]

Nou dat is grappig, ik wilde net hetzelfde stukje code hier gaan plakken.

Het resultaat is nu als volgt:
counter is: 10
counter is: 11
counter is: 12
counter is: 13
counter is: 14
counter is: 15
CountTask is onderbroken!
Klaar.

Drie dingen valt mij op tov het eerder gepostte resultaat:
1. de teller loopt nu tot 15 ipv de eerdere 14.
2. Waarom moet er een Thread.currentThread().interrupt(); stukje code achter de tweede pool.shutdownNow(); regel komen? Ik kan ook wel commentaar lezen, maar dan is mij de reden erachter onduidelijk: je wilt toch juist dat de CounterTask stopt, wat kan jou dan die hele interrupt status verder dan nog schelen, we zijn toch gestopt? Daar ging het om, toch?
3. 'Klaar.' en 'CountTask is onderbroken' nu wel in de "juiste" volgorde

Let wel op met deze code: je dient de termination tijd (=timeout) wel op 1 seconde te zetten anders heeft het hele voorbeeldtje geen effect (ik zat rustig net 60 seconden te wachten en zag de teller verder gaan dan de verwachtte 14 of 15 )

[ Voor 4% gewijzigd door redesign op 17-10-2013 17:18 ]

[Hydra]

redesign schreef op donderdag 17 oktober 2013 @ 15:53:
Het is idd netjes om de while lus te doorbreken met een false, maar soms duurt het uitwisselen van berichten te lang, en dan wil ik toch graag dat de thread om zeep geholpen wordt en netjes opgeruimd qua geclaimd geheugen.

Is dat ook mogelijk?

Het hele probleem is dat de enige nette manier de manier is die ik aangedragen heb. Vroegah had Java wel een Thread.stop() methode maar die is al sinds 't begin ongeveer deprecated omdat 'ie onveilig is: hoe een thread precies werkt is nogal platformafhankelijk. Voor meer info:
http://docs.oracle.com/ja...a/lang/Thread.html#stop()

(Zal met Android misschien iets anders werken maar daar heb ik geen ervaring mee).

[redesign]

Hydra schreef op donderdag 17 oktober 2013 @ 19:09:
[...]
hoe een thread precies werkt is nogal platformafhankelijk.

Ik denk dat hierin het probleem ligt: elk platform heeft zo zijn eigen manier van het omruimen/verwijderen van threads. Vroeger had je op je mobiel niet meerdere cpus tot je beschikking. Mijn huidige Android telefoon is krachtiger dan mijn eerste 386DX 40Mhz computer van toen

Nog even en ik krijg dit soort zinnen naar mijn hoofd geslingerd: "Opa vertel nog eens over vroeger"

[EddoH]

redesign schreef op donderdag 17 oktober 2013 @ 17:17:
...

3. 'Klaar.' en 'CountTask is onderbroken' nu wel in de "juiste" volgorde

Let wel op met deze code: je dient de termination tijd (=timeout) wel op 1 seconde te zetten anders heeft het hele voorbeeldtje geen effect (ik zat rustig net 60 seconden te wachten en zag de teller verder gaan dan de verwachtte 14 of 15 )

Even geen tijd en zin om helemaal goed je code te doorlopen maar bovenstaand doet me vermoeden dat een en ander niet werkt: timeout heet niet voor niets timeout en de functie zou volgens mij direct moeten returnen als de threads gestopt zijn. Het feit dat hij nu wel stop komt omdat je de timeout op 1 seconde hebt gezet.

Edit: ah ik zie het al, die eerste awaitTermination moet je overslaan; die wacht alleen maar tot de Threads uit zichzelf eindigen. Aangezien jij ze direct wilt sluiten dien je wel eerst shutdownNow na de shutdown aan te roepen, om vervolgens pas te wachten.

[ Voor 14% gewijzigd door EddoH op 17-10-2013 20:21 ]

[redesign]

EddoH schreef op donderdag 17 oktober 2013 @ 20:19:
[...]
Edit: ah ik zie het al, die eerste awaitTermination moet je overslaan; die wacht alleen maar tot de Threads uit zichzelf eindigen. Aangezien jij ze direct wilt sluiten dien je wel eerst shutdownNow na de shutdown aan te roepen, om vervolgens pas te wachten.

Ik heb het nu als volgt gemaakt:

Java:

public void stopCounter() {
        // laat de uitvoerende thread stoppen
        pool.shutdown(); // zorgt ervoor dat gereedstaande CounterTasks niet uitgevoerd worden
                // aangezien er geen andere CounterTasks in de queue staan, dienen we de huidige
                // CounterTask om zeep te helpen die nu wordt uitgevoerd.
        try {
            // Even wachten om te kijken of de nu uitgevoerde CounterTask er uit zichzelf mee stopt
            if (!pool.awaitTermination(1, TimeUnit.SECONDS)) {
                pool.shutdownNow(); // Na 1 seconde geloven we niet dat de huidige CounterTask
                                 // er uit zichzelf me stopt dus, breng deze dan maar helemaal om zeep.
                // We wachten nu eventjes 1 seconde om te kijken of de CounterTask
                                // daadwerkelijk de nek is omgedraaid.
                if (!pool.awaitTermination(1, TimeUnit.SECONDS))
                                        // Ohjee! CounterTask werkt niet echt mee. Hardnekkige task :)
                    System.err.println("Pool did not terminate");
            }
        } catch (InterruptedException ie) {
                       // Dit gedeelte is mij niet geheel duidelijk ????

            // (Re-)Cancel if current thread also interrupted
            pool.shutdownNow();
            // Preserve interrupt status
            Thread.currentThread().interrupt();
        }
    }

Let er wel op: de main thread is voor 4 seconden aan het wachten voordat deze de stopCounter aanroept. Dus als je de awaitTermination op 60 seconden zet, dan is de teller allang tot de 20 opgelopen!

[redesign]

Terugkomend op het eerder gepostte artikel van Brian Goetz (http://www.ibm.com/developerworks/java/library/j-jtp05236/) :

Interruption is a cooperative mechanism. When one thread interrupts another, the interrupted thread does not necessarily stop what it is doing immediately. Instead, interruption is a way of politely asking another thread to stop what it is doing if it wants to, at its convenience. Some methods, like Thread.sleep(), take this request seriously, but methods are not required to pay attention to interruption. Methods that do not block but that still may take a long time to execute can respect requests for interruption by polling the interrupted status and return early if interrupted. You are free to ignore an interruption request, but doing so may compromise responsiveness.

Dus eigenlijk wordt men vriendelijk doch met klem gevraagd om vooral in de eigen code to pollen of de interrupted status flag is gezet, en dan zelf de conclusie te trekken - wanneer dat kan - om het bijltje er bij neer te gooien.

In code zou zoiets er dan (naar mijn idee) zo uit moeten zien:

Java:

public void run() {
    while(!cancelled) {
          if (!cancelled) {
             blockingCallOne();
          }
          if (!cancelled) {
             blockingCallTwo();
          }
          if (!cancelled) {
             blockingCallThree();
          }
    } // einde while lus

    public void blockingCallOne() throws InterruptedException {
        // 1. doe blokkerende dingen op dit punt
        try {
           Thread.sleep(1000);
        } catch (InterruptedException e) {
          cancelled = true;
          Thread.currentThread().interrupt();
       }
    }

    public void blockingCallTwo() throws InterruptedException {
        // 2. doe blokkerende dingen op dit punt
        try {
           Thread.sleep(1000);
        } catch (InterruptedException e) {
          cancelled = true;
          Thread.currentThread().interrupt();
       }
    }

    public void blockingCallThree() throws InterruptedException {
        // 3. doe blokkerende dingen op dit punt
        try {
           Thread.sleep(1000);
        } catch (InterruptedException e) {
          cancelled = true;
          Thread.currentThread().interrupt();
       }
    }

}

Op die manier zou je dus je hele-lange-blocking-method kunnen opsplitsen in een dezelfde method maar die dan wel veel beter reageert op interrupts van buitenaf. Het werk blijft hetzelfde, alleen de "responsiveness" is verbeterd. Althans, dat is mijn theorie

Iemand andere/betere ideeen?

Edit: try-catch in elke blocking call geplaatst om DAAR de boolean te zetten ipv in de while lus.

[ Voor 15% gewijzigd door redesign op 17-10-2013 21:50 ]