[Thread] Executor et timeout

Version imprimable

19/09/2012, 09h56
BiM

[Thread] Executor et timeout
Bonjour,

J'ai cherché sur le forum sans succès sur le cas précis. Bon, il y a peut être la réponse quelque part, mais je ne l'ai pas trouvée.

J'ai un executor qui exécute une liste de tâches (Runnable). Je veux qu'au bout de 2h d’exécution de la liste, l'exécution s'arrête.

Je me demandais donc :
- Peut-on rajouter un timeout à l'executor ? (Et non à chaque tâche lancée)
- Ou alors, peut-on lancer une méthode qui se déclenchera dans 2h si l'exécution n'est pas terminée ? Et qu'on peut interrompre si l'exécution est terminée avant les 2h ?
La deuxième solution me semble faisable mais je ne fais pas ça tous les jours alors c'est pas vraiment ma spécialité.

J'attends un peu de votre aide. Merci !
19/09/2012, 10h52
wax78

Je sais que dans, par exemple, ThreadPoolExecutor, il y'a une méthode qui permet d'attendre les thread avec une timeout :

JavaDoc
19/09/2012, 10h59
BiM

Mais j'ai cru comprendre que cette méthode permet de déclencher la fermeture des threads en cours, s'ils ne sont pas fermés au bout de X temps, tant pis pour eux.

Donc cela déclenche dès le début la fermeture de ceux-ci non ?

NB : J'ai justement un ExecutorService en entrée.
19/09/2012, 13h05
wax78

Non, pour moi, cela attends soit la fin des threads lancés, soit le timeout s'ils ne sont pas terminés.
19/09/2012, 14h16
BiM

Citation:

Envoyé par wax78

Non, pour moi, cela attends soit la fin des threads lancés, soit le timeout s'ils ne sont pas terminés.

Ah OK, du coup c'est bien ce que je cherche mais j'arrive pas à m'en servir. En fait, je veux l'intégrer dans une application existante et l'implémentation m'a l'air bien complexe.

Est-ce que tu peux me donner un exemple d'utilisation de awaitTermination() sans qu'elle soit précédée d'un shutdown() stp ?

La méthode execute n'est pas bloquante normalement ?
19/09/2012, 14h37
wax78

Pourquoi ne veux tu pas mettre le shutdown ? Le shutdown ne terminera pas brutalement les thread qui sont dans la file et qui doivent se derouler et se terminer.

En gros tu submit les taches. Tu shutdown puis tu AwaitTermination avec un timeout.

mais j'oublie peut être une chose ou je n'ai pas compris exactemetn ce que tu désire ;)
19/09/2012, 15h10
BiM
J'ai une liste de tâches à exécuter, si ces tâches ne sont pas terminées dans les 2h, je veux qu'elles s'arrêtent (tant pis pour elles). Ca c'est pour le principe.

A noter que c'est un CRON trigger qui vient exécuter toutes les nuits les tâches à minuit.

Le mieux c'est qu'à 2h du matin, si l'ensemble des tâches ne sont pas terminées, on les interrompt (pour libérer les ressources du serveur).

Le CRON lance un executor qui contient une liste de tâches.

J'ai plusieurs solutions mais je n'arrive à en implémenter aucune (tout ce qui est scheduler et autres c'est pas ma tasse de thé) :
- Lancer une commande avec un timer qui interrompt l'executor au bout de 2h.
- Mettre un timeout sur l'executor.
- Mettre un timeout sur chaque tâche en fonction du temps qu'il reste (Je décompte le temps à chaque fois qu'une tâche est lancée).
- Mettre un timeout sur le CRON.
- Déclencher un CRON à 2h du matin qui interrompt le premier CRON s'il n'est pas fini (c'est tordu ça) mais ça semble être la meilleure solution (en espérant qu'on puisse interrompre un Job CRON comme ça).
Je ne comprends pas pourquoi utiliser un shutdown() dans mon cas avant le awaitTermination(), ou alors je n'ai pas bien compris l'utilité de chacun. J'ai cru comprendre que le shutdown() laissais la tâche en cours se finir mais empêche les autres tâches de la liste de se dérouler. Le awaitTermination() qui suit permet d'attendre X temps avant que la tâche en cours se finisse. Si elle ne s'est pas finie à temps il renvoie false, on peut alors faire d'autres action : par exemple faire un shutdownNow() pour forcer l'arrêt de la tâche en cours.

J'ai peut être tout faux.

EDIT : Dans l'implémentation actuelle, on utilise jamais submit mais execute, enfin, j'y comprends pas grand chose est faire un copier/coller des sources ici c'est pas l'idéal, faudrait que je les simplifie un peu avant d'abord.
19/09/2012, 15h20
wax78

Es-tu sûre d'avoir bien lu la javadoc ? http://docs.oracle.com/javase/6/docs...orService.html

shutdown() : Initiates an orderly shutdown in which previously submitted tasks are executed, but no new tasks will be accepted.
awaitTermination(long timeout, TimeUnit unit) :Blocks until all tasks have completed execution after a shutdown request, or the timeout occurs, or the current thread is interrupted, whichever happens first.

Pour moi, après le shutdown, tu ne pourras plus submit des taches. Mais toutes les tâches de la file seront executées.

Tu devrais essayer avec un exemple bidon avec un sleep dans les thread et regarder le comportement, en 20 lignes c'est faisable.

Edit : Je dis parfois des bêtises aussi personnes n'est infaillible ;)

Mais moi aussi je parle de ThreadPoolExecutor par exemple ce qui change peut etre les choses niveau du submit/execute.
19/09/2012, 15h43
BiM

Ok merci.

Ben j'ai bien lu la doc, mais je crois que j'ai besoin de reprendre des cours d'anglais ;)

Bon, je vais essayer de faire un test indépendant avec une implémentation qui ressemble à ce que j'ai pour pouvoir me faire aider à nouveau (à moins que je ne réussisse toute seule ;)).

Merci à toi en tout cas, je reviens ici pour dire si mon sujet est résolu ou a besoin d'être approfondi.
19/09/2012, 16h00
wax78

Sans problèmes, bonne chances et a bientôt (ou pas)
19/09/2012, 16h08
tchize_

De toutes façons, on ne peux pas interromptre un thread en java (les méthodes associées sont dépréciées). Du coup, si tas une tâche qui met 4 heures à s'exécuter, ben ton executor attendra au minimum ce temps là. Pas glop.

Si la contrainte est que ton application doit libérer la place quoi qu'il arrive au bout de 2h, je ferais un simple timer au début de l'application qui appellerais System.exit(-1) au bout de 2h :)
19/09/2012, 16h10
wax78

C'est effectivement une très bonne alternative qui coutera pas cher.:ccool:
19/09/2012, 16h17
BiM

Heu... mais ça tue l'ensemble de l'application ça ? Le Thread est sur une application qui ne s'arrête pas. On ne peut vraiment pas tuer un Thread ?

EDIT : La méthode interrupt(), elle sert bien à ça non ?
19/09/2012, 16h30
wax78

Cela va dépendre des opérations effectuées dans le Thread, certains opérations ne sont pas interruptibles et donc la effectivement interrupt ne servira a rien.

Mais par contre, si tu programme toi même le contenu du thread et que par exemple c'est une longue boucle qui ne fait que des additions de chiffre, tu pourrais t'arranger pour rendre celle ci interruptible facilement mais ce n'est pas toujours le cas malheureusement :(
19/09/2012, 16h54
BiM

J'ai effectivement la main sur le code des Thread.

En gros, c'est de l'écriture dans des fichiers CSV. Celui qui sera interrompu ne sera de toute façon pas valide. Mais je peux peut être me débrouiller pour qu'il finisse quand même proprement. Genre la fin de la ligne qu'il est en train d'écrire. Mais je ne vois pas trop comment faire ça.
19/09/2012, 21h24
balsoc

Bonjour,

Pour ce genre de tâche lourde, une solution est de faire une abstract class sous l'interface Runnable qui gère des cycles au lieu d'un run complet.
Simplement avec un flag, et un éventuellement un délai.

La méthode run() boucle sur le flag et appelle une méthode abstraite.

Au final les classes effectives implémentent la gestion "streamée" sans devoir gérer l'exécution.

Et avec ça tu peux facilement les stopper dans un délai raisonnable (ou du moins estimable).

Pour avoir un lien direct avec l'instance de ExecutorService, j'imagine que faire la classe abstraite en dessous de Thread est une meilleure solution.

D'ailleurs la méthode shutdownNow est décrite comme ceci:

Citation:

There are no guarantees beyond best-effort attempts to stop processing actively executing tasks. For example, typical implementations will cancel via Thread.interrupt(), so if any tasks mask or fail to respond to interrupts, they may never terminate.

Donc typiquement, si dans la boucle principale tu fais

Code:

while(flag && !isInterrupted())

, tu devrais coller au machin.

(Mais ça je n'ai jamais testé)

[EDIT]En fait tu peux faire Thread.currentThread().isInterrupted() dans le Runnable donc c'est nettement plus propre.[/EDIT]
19/09/2012, 22h14
tchize_

Citation:

Envoyé par BiM

Heu... mais ça tue l'ensemble de l'application ça ? Le Thread est sur une application qui ne s'arrête pas. On ne peut vraiment pas tuer un Thread ?

Ha bé tu parle de libérer les ressource,s j'ai supposé qu'on parlais de la mémoire :)

Pour arrêter un thread, il faut sa collaboration. Le interrupt() sert à arrêter tout opération blocante (typiquement les IOs) avec un InterruptedException. Mais ça ne suffit pas, il faut aussi organiser un système de flag dans ton thread que tu passe à true pour dire "ho la, stop" et ton code, dans les opérations potentiellements longues et répétitives, devra en tenir compte.

Et ça ne résoud pas encore le problème du exécutor, mais tu peux mettre une flag global et avoir tous les threads du executor checker ce flag avant de démarrer :)

En fait, c'est libérer la lecture et l'écriture du disque, la base de données, etc.

Bon j'ai pas tout compris sur la notion du interrupt(). En réalité, pour l'instant, j'ai des Runnable et non des Thread. Du coup, je n'ai pas de méthode interrupt() (Enfin s'il faut changer ça, ça peut se faire, mais quelle est la dif entre Thread et Runnable dans ce cas ?). Les tâches ne sont pas lancées en parallèle, c'est important comme notion, si c'était le cas, tout échouerait.

Toutes mes tâches ont une classe mère, donc je peux m'en servir pour rajouter quelques notions. En gros, on a une méthode qui exécute une requête SQL et qui fetch chaque résultat. Je voudrais que l'opération ne puisse pas être interrompue au milieu d'un fetch mais seulement avant ou après celui-ci.

J'ai quelque chose comme ça dans mon Runnable :
Code:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 OutputStream outputStream = null; try { outputStream = new BufferedOutputStream(new FileOutputStream(getExportFile(getExportFileName()))); // Write lines one by one while (rs.next()) { IOUtils.write(fetchResult(rs), outputStream); } } catch (SQLException e) { getLogger().error("SQL Exception", e); } catch (FileNotFoundException e) { getLogger().error("Cannot find the file", e); } catch (IOException e) { getLogger().error("IOException while writing file", e); } finally { IOUtils.closeQuietly(outputStream); }
Comment feriez-vous quelque chose de propre au moment de l'interruption ?

Dernière question, une fois qu'on a fait un shutdown sur l'Executor, est-il possible de le redémarrer ou il faut en instancier un tout beau tout neuf ?

20/09/2012, 10h17
adiGuba
Citation:

Envoyé par BiM

Bon j'ai pas tout compris sur la notion du interrupt().

Il ne faut pas utiliser stop(), car cela va tuer le thread "brutalement", ce qui risque de laisser l'application dans un état incohérent (données partiellement modifié, etc.).

Du coup interrupt() ne fait rien, mis à part modifier l'état du thread (que tu peux vérifier via Thread.interrupted() ou Thread.currentThread().isInterrupted().

En fait cela permet de dire au thread : "tu dois t'arrêter".
Mais pour cela le thread doit vérifier son état "interrupted" régulièrement, et le prendre en compte...

Citation:

Envoyé par BiM

En réalité, pour l'instant, j'ai des Runnable et non des Thread. Du coup, je n'ai pas de méthode interrupt() (Enfin s'il faut changer ça, ça peut se faire, mais quelle est la dif entre Thread et Runnable dans ce cas ?).

Le Thread représente plus le thread d'exécution, tandis que le Runnable correspond à la tâche à exécuter.

De toute manière si tu utilises un Executors tu n'as pas besoin de manipuler le thread puisqu'il est caché. Par contre tu peux l'interrompre via l'objet Future<?> que tu récupère après le submit de ta tâche.

Un future.cancel(true) génèrera un interrupt si le thread n'est pas terminé.

Au sein de ton Runnable, tu peux utiliser Thread.currentThread() pour récupérer le thread et vérifier son status "interrupted"...

Citation:

Envoyé par BiM

Toutes mes tâches ont une classe mère, donc je peux m'en servir pour rajouter quelques notions. En gros, on a une méthode qui exécute une requête SQL et qui fetch chaque résultat. Je voudrais que l'opération ne puisse pas être interrompue au milieu d'un fetch mais seulement avant ou après celui-ci.

Aucun problème justement : c'est à toi de voir où tu vas positionner le code qui va vérifier le status "interrupt", et donc où la tâche sera interrompue...

Par contre il faut juste savoir que certains méthodes sont déjà interruptible, et génère une exception lorsqu'elle reçoive un interrupt (c'est le cas de sleep() par exemple).

Mais en gros dans ton code, tu peux faire quelque chose comme cela :
Code:

1 2 3 4 5 public static void checkStatus() { if (Thread.interrupted()) { throw new RuntimeException("interrupted"); } }
Puis il ne te restera plus qu'à appeller "checkStatus()" à divers endroits...
Tu peux même personnaliser l'exception, afin de mieux la logger par exemple.

Citation:

Envoyé par BiM

J'ai quelque chose comme ça dans mon Runnable :

Juste une remarque : Perso je n'aime pas trop les try/catch/finally.
Je préfère largement les try/finally englobé dans un try/catch. C'est plus propre (on n'ignore pas les exceptions de fermeture du flux).

a++

Tu peux faire quelquechose comme cela:

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
        ExecutorService executor = ...;
        Collection<Callable<Object>> tasks = ...;
 
        // Execute et attend la fin ou les 2 heures
        List<Future<Object>> futures = executor.invokeAll(tasks, 2, TimeUnit.HOURS);
 
        // Là, tout et fini ou les 2h se sont écoulées
        for (Future<Object> future : futures)
        {
            // On annule toutes les tâches non terminées, avec interruption si nécessaire
            if (!future.isDone())
            {
                // La tâche ne sera pas lancée
                future.cancel(true);
            }
        }
 
        // on achève l'executor pour libérer les resources
        executor.shutdown();

Il reste ensuite à gérer les interruptions dans tes tâches.

Note: Je soupçonne la 2e partie de mon code d'être équivalente à un appel à shutdownNow().

EDIT: grillé :aie:

20/09/2012, 10h19
tchize_

Citation:

Envoyé par BiM

Bon j'ai pas tout compris sur la notion du interrupt().

interrupt() met le flag isInterrupted() du thread à true et interrompt immédiatement toute opération bloquante, typiquement de type IO, réseau ou sleep le supportant (typiquement, toutes les méthodes lancant un interruptedException).

Citation:

quelle est la dif entre Thread et Runnable

Runnable est juste une interface contenant une méthode run. Thread est une classe permettant la création d'un thread et qui, lors du démarrage de ce thread, lance la méthode run d'un Runnable qui lui a été passé ou de lui même.

Citation:

Les tâches ne sont pas lancées en parallèle, c'est important comme notion, si c'était le cas, tout échouerait.

Alors pourquoi tu va t'emmerder avec un executorService au lieu de faire une simple boucle sur toutes tes tâches?

Citation:

Je voudrais que l'opération ne puisse pas être interrompue au milieu d'un fetch mais seulement avant ou après celui-ci.

Ben tu met tes tests où tu veux hein :mouarf: Par contre si tu ne veux pas arrêter au milieu d'un fetch -> tu oublie le interrupt(), car typiquement il pourrais interrompre un IO réseau à l'intérrieur de JDBC! Tu met juste une méthode type isTooLate() dans le parent qui va te renvoyer true ou false suivant l'heure ^^ . Méthode que tu appelle sagement entre chaque opération courte. Bien sur, ça veux dire qu'en cas d'interruption réseau prolongée, de lock de la DB, etc, ta tâche pourrais dépasser les 2h juste parce qu'elle attends 30 minutes une réponde de la DB ;)

Citation:

Dernière question, une fois qu'on a fait un shutdown sur l'Executor, est-il possible de le redémarrer ou il faut en instancier un tout beau tout neuf ?

Faire un nouveau, mais le executor c'est pour du paralléllisme qu'on utilise ça. Si t'as pas de parallélisme -> tu oublie, ça sert à rien d'en mettre pour le plaisir d'en mettre. T'as typiquement un batch d'opération qui doit s'interrompre à un moment. Ben tu met tout ça en série dans une boucle avec un check régulire sur l'heure ^^

Code:

1
2
3
4
5
// Write lines one by one
while (!Thread.currentThread().isInterrupted() && rs.next())
{
        IOUtils.write(fetchResult(rs), outputStream);
}

Le shutdown du pool appellera interrupt sur chaque thread.

20/09/2012, 10h33
adiGuba

Citation:

Envoyé par tchize_

Ben tu met tes tests où tu veux hein :mouarf: Par contre si tu ne veux pas arrêter au milieu d'un fetch -> tu oublie le interrupt(), car typiquement il pourrais interrompre un IO réseau à l'intérrieur de JDBC!

Oui en effet je n'avais pas pensé à cela !

Citation:

Envoyé par tchize_

Tu met juste une méthode type isTooLate() dans le parent qui va te renvoyer true ou false suivant l'heure ^^ . Méthode que tu appelle sagement entre chaque opération courte.

Et surtout sans oublier de mettre ce champ en "volatile" ;)

a++
20/09/2012, 13h57
tchize_

Citation:

Envoyé par adiGuba

Oui en effet je n'avais pas pensé à cela !

Et surtout sans oublier de mettre ce champ en "volatile" ;)

Pas besoin si la méthode calcule elle même le temps et qu'il n'y a pas d'autre thread :)

Merci pour vos réponses, vous battez pas :aie:

Ces éclaircissements m'aident beaucoup ;)

Citation:

Envoyé par tchize_

Alors pourquoi tu va t'emmerder avec un executorService au lieu de faire une simple boucle sur toutes tes tâches?

J'utilise un Executor parce que c'était un Executor à la base et que j'ai pas envie de tout casser mais j'utilise cet Executor : executors.newSingleThreadExecutor(), ce qui me permet de les mettre dans la file d'exécution mais qu'ils s’exécutent les uns après les autres.

Voici à quoi ressemble mon Thread (Runnable plutôt), un truc très simple de test :
Code:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 public class MyThread implements Runnable { private String _valeur; public MyThread(String valeur) { _valeur = valeur; } public String getValeur() { return _valeur; } @Override public void run() { try { System.out.println("Début " + _valeur); Thread.sleep(2000); System.out.println("Fin " + _valeur); } catch (InterruptedException e) { System.err.println("Erreur " + _valeur); } } }
Je gère le executor comme suit :
Code:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 ExecutorService executor = Executors.newSingleThreadExecutor(); executor.submit(new MyThread("Anne")); executor.submit(new MyThread("Bertrand")); executor.submit(new MyThread("Camille")); executor.submit(new MyThread("Diane")); executor.submit(new MyThread("Eric")); executor.submit(new MyThread("Fabien")); executor.submit(new MyThread("Gertrude")); executor.submit(new MyThread("Hermione")); executor.submit(new MyThread("Isabelle")); executor.submit(new MyThread("Julien")); executor.shutdown(); if (!executor.awaitTermination(11, TimeUnit.SECONDS)) { System.out.println("Not finished"); executor.shutdownNow(); } else { System.out.println("Finished"); }
J'obtiens ça :

Citation:

Début Anne
Fin Anne
Début Bertrand
Not finished
Erreur Bertrand

Donc lors du shutdownNow() qui est censé appelé les méthodes interrupt() des Thread en cours, le thread en question se prend un InterruptedException dans sa face :aie: Le thread ne continue donc pas son boulot tranquillement.

Citation:

Envoyé par adiGuba

Juste une remarque : Perso je n'aime pas trop les try/catch/finally.
Je préfère largement les try/finally englobé dans un try/catch. C'est plus propre (on n'ignore pas les exceptions de fermeture du flux).

Vu le nombre de couches dans l'application, on va laisser comme ça, ça tourne très bien.
Citation:
Envoyé par Deaf
Tu peux faire quelquechose comme cela:
Code:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 ExecutorService executor = ...; Collection<Callable<Object>> tasks = ...; // Execute et attend la fin ou les 2 heures List<Future<Object>> futures = executor.invokeAll(tasks, 2, TimeUnit.HOURS); // Là, tout et fini ou les 2h se sont écoulées for (Future<Object> future : futures) { // On annule toutes les tâches non terminées, avec interruption si nécessaire if (!future.isDone()) { // La tâche ne sera pas lancée future.cancel(true); } } // on achève l'executor pour libérer les resources executor.shutdown();
Intéressant. Je vais tenter ce type de solution. Mais le problème des interrupt est toujours le même. Il s'avère que invokeAll correspond quand même carrément à mon besoin (j'en ai l'impression), merci :ccool:

Citation:

Envoyé par Deaf

Note: Je soupçonne la 2e partie de mon code d'être équivalente à un appel à shutdownNow().

Après analyse du shutdown de l'Executor, il interrompt tous les Thread en cours par un interrupt().

Citation:

Envoyé par tchize_

interrupt() met le flag isInterrupted() du thread à true et interrompt immédiatement toute opération bloquante, typiquement de type IO, réseau ou sleep le supportant (typiquement, toutes les méthodes lancant un interruptedException).

Ca doit être pour ça que j'ai une Exception ci-dessus, mais ça risque d'être pareil lors de l'écriture dans un fichier, non ?
Citation:
Envoyé par balsoc
Code:

1 2 3 4 5 // Write lines one by one while (!Thread.currentThread().isInterrupted() && rs.next()) { IOUtils.write(fetchResult(rs), outputStream); }
Le shutdown du pool appellera interrupt sur chaque thread.
C'est bien ce qui me semblait, mais bon toujours ce problème d'IO à priori.

21/09/2012, 00h20
tchize_

Citation:

Envoyé par BiM

Ca doit être pour ça que j'ai une Exception ci-dessus, mais ça risque d'être pareil lors de l'écriture dans un fichier, non ?

Ben oui, c'est le but de interrupt. Comme je l'ai dit, si tu veux pouvoir finir ta transaction DB, ta lecture ou cloturer ton fichier, l'interrupt n'est pas le bon choix. Du coup l'exécutor n'est plus le bon choix non plus :aie:
21/09/2012, 08h19
BiM

Bon disons que c'est pas grave si je finis pas d'écrire ma ligne. Au pire, je me choppe un InterruptedException, et dans le finally, je ferme proprement le fichier (ou la connexion à la DB selon l'emplacement où je me trouve), c'est ça ?
21/09/2012, 13h47
tchize_

ben oui, c'est toi qui disais que tu voulais terminer ton fetch.
21/09/2012, 14h49
BiM

Citation:

Envoyé par tchize_

ben oui, c'est toi qui disais que tu voulais terminer ton fetch.

Oui mais au final je me rends compte qu'il y a trop de choses à modifier.