Parallélisme régulé

Objectifs

Prérequis

Vous devez savoir parfaitement comment définir une activité (Thread) en Java, comment lancer une activité, et comment attendre sa terminaison.

** Si ce n'est pas le cas, (re)voyez et (re)faites le travail demandé à la rubrique « concurrence et cohérence » avant d'entamer ce TP**

Vous aurez vraisemblablement besoin lors de ce TP d'utiliser les méthodes de classe suivantes de la classe Thread :

Enfin, vous aurez sans doute aussi besoin de deux méthodes de classe de la classe System : System.nanoTime() et System.currentTimeMillis() qui fournissent une durée écoulée (en ns et ms) depuis une date d'origine non spécifiée. La différence entre les valeurs retournées par deux appels successifs permet d'évaluer le temps écoulé entre ces deux appels.

Préparation : services de régulation des activités en Java

La rapide présentation qui suit peut être complétée par la lecture de la partie correspondante du cours sur les processus légers (planches 34-45) pour les notions et sur la documentation Java en ligne pour la syntaxe et les détails techniques.

Les classes et notions utilisées jusqu'ici étaient destinées à définir et gérer la concurrence explicitement, et à un niveau fin : le choix de lancer, d'attendre et de terminer une tâche appartient entièrement au programmeur. De même, le programmeur a la charge des choix en termes de gestion de la cohérence (variables volatile, classes atomiques...) et du type d'attente (blocs synchronized, verrous, attente active).

La plateforme Java fournit dans ses dernières versions la classe Executor, destinée à séparer la gestion des activités des aspects purement applicatifs. Le principe est qu'un objet de la classe Executor (« exécuteur ») fournit un service de gestion et d'ordonnancement d'activités, auquel on soumet des tâches à traiter. Une application est donc vue comme un ensemble de tâches qui sont fournies à l'exécuteur. L'exécuteur gère alors l'exécution des tâches qui lui sont soumises de manière indépendante et transparente pour l'application. L'objectif de ce service est de permettre

Le paquetage java.util.concurrent définit 3 interfaces pour les exécuteurs :

Le paquetage java.util.concurrent fournit différentes implémentations d'exécuteurs. Le principe commun aux exécuteurs est de distribuer les tâches soumises à un ensemble d'ouvriers. Chaque ouvrier est un thread cyclique, qui traite une par une les tâches qui lui sont attribuées.

Les exécuteurs fournis par le paquetage java.util.concurrent sont de deux sortes :

Pools de threads

La classe java.util.concurrent.Executors fournit des méthodes permettant de créer des pools de threads implantant ExecutorService avec un nombre d'ouvriers fixe -- méthode newFixedThreadPool --, variable (adaptable) -- méthode newCachedThreadPool) ou permettant une régulation par vol de tâches (voir cours) (méthode newWorkStealingPool). Une variante implantant ScheduledExecutorService est proposée pour chacune de ces méthodes, afin de permettre d'intervenir sur l'ordonnancement des tâches. Enfin, les classes java.util.concurrent.ThreadPoolExecutor et java.util.concurrent.ScheduledThreadPoolExecutor proposent encore davantage d'options sur la paramétrage et la supervision de l'ordonnancement.

Les pools de threads évitent la création de nouveaux threads pour chaque tâche à traiter, puisque qu'un même ouvrier est réutilisé pour traiter une suite de tâches, ce qui présente plusieurs avantages :

D'une manière générale,

Classes et méthodes utiles
Un exemple
import java.util.concurrent.Future;
import java.util.concurrent.Callable;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutorService;

class SigmaC implements Callable<Long> {
    private long début;
    private long fin;

    SigmaC(long d, long f) { début = d; fin = f;}

    @Override
    public  Long call() { // le résultat doit être un objet
        long s = 0;
        for (long i = début; i <= fin; i++) s = s + i;
        return s;
    }                
}

class SigmaR implements Runnable {
    private long début;
    private long fin;

    SigmaR(long d, long f) { début = d; fin = f;}

    @Override
    public  void run() {
        long s = 0;
        for (long i = début; i <= fin; i++) s = s + i;
        System.out.println("Calcul terminé. ∑("+début+","+fin+") = "+s);
    }                
}

public class Somme {     
    public static void main(String[] args) throws Exception {
    
        ExecutorService poule = Executors.newFixedThreadPool(2);
           
        Future<Long> f1 = poule.submit(new SigmaC(0L,1_000_000_000L));
        Future<Long> f2 = poule.submit(new SigmaC(0L,4_000_000_000L));
        poule.execute(new SigmaR(900_000L,1_000_000_000L));
        Future<Long> f3 = poule.submit(new SigmaC(1,100));
        Future<Long> f4 = poule.submit(new SigmaC(0L,3_000_000_000L));
    
        poule.shutdown();
    
        System.out.println("Résultat obtenu. f1 = "+f1.get());
        System.out.println("Résultat obtenu. f2 = "+f2.get());        
        System.out.println("Résultat obtenu. f3 = "+f3.get());        
        System.out.println("Résultat obtenu. f4 = "+f4.get());
    }    
}
Commentaires

Pool Fork/Join (Schéma Map/Reduce)

La classe ForkJoinPool est un exécuteur dont l'ordonnancement est adapté à une parallélisation selon le schéma fork/join (voir cours, planches 43-45). Le principe (récursif) est

Ce schéma de programmation permet de créer dynamiquement un nombre de tâches adapté à la taille de chacun des (sous)-problèmes rencontrés, chacune des tâches créées représentant une charge de travail équivalente. Ce schéma est donc bien adapté au traitement de problèmes irréguliers, de grande taille. L'ordonnanceur de la classe ForkJoinPool comporte en outre une régulation (vol de tâches) qui permet l'adaptation de l'exécution aux capacités de calcul disponibles.

Il est important de noter que ce schéma repose sur le fait que les sous-tâches créées s'exécutent en parallèle, et donc sur l'hypothèse qu'elles sont complètement indépendantes. Tout conflit d'accès aux ressources, ou synchronisation compromet l'efficacité de ce schéma. Le schéma Fork/Join est donc idéalement et principalement destiné aux calculs intensifs, irréguliers, en mémoire pure (sans E/S). Avec ce schéma, les interactions et synchronisations entre tâches sont alors limitées aux interactions entre une tâche mère et ses tâches filles, lorsque celles-ci ont terminé, et que la tâche mêre récupère les résultats des tâches filles pour les agréger.

Classes et méthodes utiles
Un exemple

(fourni également dans l'archive jointe) réalise le schéma fork/join et illustre l'utilisation des principales classes et méthodes dans ce cadre. Dans cette application, les données à traiter sont représentées par un simple entier, qui symbolise leur volume.

import java.util.concurrent.RecursiveTask;
import java.util.concurrent.ForkJoinPool;

class TraiterProblème extends RecursiveTask<Integer> {

 private int resteAFaire = 0;
 private int résultat = 0;
 static final int SEUIL = 10;

 TraiterProblème(int resteAFaire) {
    this.resteAFaire = resteAFaire;
}

 protected Integer compute() {

    //si la tâche est trop grosse, on la décompose en 2
    if(this.resteAFaire > SEUIL) {
        System.out.println("Décomposition de resteAFaire : " + this.resteAFaire);

        TraiterProblème sp1 = new TraiterProblème(this.resteAFaire / 2);
        TraiterProblème sp2 = new TraiterProblème(this.resteAFaire / 2);

        sp1.fork();
        sp2.fork();

        résultat = sp1.join()+ sp2.join();

        return résultat;

    } else {
        System.out.println("Traitement direct de resteAFaire : " + this.resteAFaire);
        return resteAFaire * 3;
    }
 }
}

public class FJG {
 static ForkJoinPool fjp = new ForkJoinPool();
 static final int TAILLE = 1024; //Attention : nécessairement une puissance de 2

 public static void main(String[] args) throws Exception {
    TraiterProblème monProblème = new TraiterProblème(TAILLE);
    int résultat = fjp.invoke(monProblème);
    System.out.println("Résultat final = " + résultat);
 }
}
Commentaires
Quelques écueils

Exercices

Vous aurez vraisemblablement besoin pour cette partie de deux méthodes de classe de la classe System : System.nanoTime() et System.currentTimeMillis() qui fournissent une durée écoulée (en ns et ms) depuis une date d'origine non spécifiée. La différence entre les valeurs retournées par deux appels successifs permet d'évaluer le temps écoulé entre ces deux appels.

L'archive fournie propose différents exercices.
Chaque exercice comporte un calcul séquentiel (itératif ou récursif), qu'il faut paralléliser en utilisant un pool fixe et/ou un pool Fork/Join.
Chaque exercice comporte une méthode main permettant de lancer et comparer les différentes versions. Des commentaires // ********* A compléter ou // ********* A corriger signalent les (seuls) endroits du code où vous devez intervenir pour implanter les versions parallèles du calcul séquentiel fourni.

Les exercices utilisent des tableaux d'entiers stockés sur disque.
L'archive fournie comporte une application GCVT.java qui propose une classe TableauxDisque permettant de générer, charger en mémoire, sauvegarder ou comparer de tels tableaux.
La méthode mainde l'application GCVT.java permet en outre d'appeler les méthodes de la classe TableauxDisque depuis la console.
Cette application pourra en particulier être utilisée pour générer les jeux de données utiles aux tests. En effet, pour que le gain apporté par les versions parallèles soit sensible, il est nécessaire que les volumes de données traités soient significatifs, ce qui implique ici de travailler (pour l'évaluation de performances) sur des tableaux de 1 à 100 millions d'entrées, ce qui aurait alourdi inutilement l'archive. Vous devrez donc générer vos jeux de données avec cette application, sans oublier de supprimer les fichiers créés une fois le TP passé, sans quoi vous risquez d'épuiser votre quota d'espace disque :)

Les exercices peuvent être traités dans l'ordre suivant :

Tester les performances d'applications concurrentes en Java : quelques remarques pratiques