Commençons par nous intéresser aux classes héritant de ForkJoinTask. Cette dernière propose une base aux tâches amenées à être parallélisées. Une tâche ainsi définie recoupe un peu la notion de Callable apportée par Java 5.

Prenons un exemple simple, souvent utilisé, concernant la résolution de la suite de Fibonacci (vous n'aimez pas les Maths ? Ne fuyez pas déjà ! L'exemple est simple :) ).

Il s'agit de résoudre, de façon récursive (*), la suite suivante :

f(n) = f(n-1) + (f(n-2) pour n > 1 avec f(0) = 0; f(1) = 1;

Ce qui donne pour les premiers entiers : 0, 1, 1, 2, 3, 5, 8, 13, 21, 34, ...

(*) : Vous trouverez partout un algo beaucoup plus puissant pour cette résolution, mais non récursif et donc pas intéressant dans le cadre de notre démonstration.

L'approche naturelle ici est de faire calculer chaque partie de la suite en parallèle de l'autre (f(n-1) serait calculé en parallèle de f(n-2).

Nous allons donc créer une classe étendant ForkJoinTask implémentant cette logique. Nous considérerons la classe intermédiaire RecursiveAction (étendant ForkJoinTask) et se prêtant justement aux opérations récursives.

Voici un exemple d'implémentation :

public class Fibonacci extends RecursiveAction {

      private final int number;

      private int result;

      public Fibonacci(int number) {

            this.number = number;

      }

      public int getResult() {

            return result;

      }

      public void compute() {

            int n = number;

            if (n < 2) {

                  result = n;

                  return;

            }

            final Fibonacci f1 = new Fibonacci(n - 1); // subdivision

            final Fibonacci f2 = new Fibonacci(n - 2); // en 2 sous taches

            forkJoin(f1, f2); // invocation en parallèle des sous-taches (fork & join)

            result = f1.getResult() + f2.getResult(); // agregation des resultats

      }

}

 Pour exécuter ce code, nous y ajoutons une méthode main() :

      public static void main(String[] args) {

            final ForkJoinPool pool = new ForkJoinPool(2);

// dual core, mono processeur : pas la peine de tailler le pool au dela

            Fibonacci f = new Fibonacci(44);

// 44 est une valeur interessante pour son temps de calcul exploitable

            pool.invoke(f);

            System.out.println(f.getResult());

      }  

Cette méthode comporte la déclaration d’un ForkJoinPool, pierre angulaire ici car responsable de la gestion des Threads liés à l’exécution. Le paramètre passé au constructeur n’est autre que la taille du pool. On veillera à fournir une taille inférieure ou égale au nombre d’unité de traitement disponibles sur le serveur. Une valeur supérieure n’apporterait rien : les threads devant attendre la disponibilité d’une unité de calcul pour y être exécutés.

En lançant cet exemple, on observe une occupation CPU totale pour peu que le paramètre passé au ForkJoinPool soit égal au nombre de cores total sur le serveur.

Maintenant comparons avec une exécution récursive mais non parallèle en ajoutant une méthode à notre classe :

[...]

      private int seqFib(int n) {

            if (n <= 1) {

                  return n;

            } else {

                  return seqFib(n - 1) + seqFib(n - 2);

            }

      }

      public void sequentialCompute() {

            result = seqFib(this.number);

      }

[...]

L'heure de la comparaison sonne ! Et là, c'est plutôt une grosse déception car l'approche classique est particulièrement plus rapide que l'approche parallélisée. 

Essayons de comprendre pourquoi :

En essayant de paralléliser le traitement, nous découpons le problème en tâches très petites car tant que n n'est pas inférieur (strict) à 2, deux nouvelles tâches sont créées.

Cette création a déjà un coût qu’on ne retrouve pas dans l'autre approche. Ensuite, les tâches vont être dépilées tour à tour (en parallèle) pour agréger les résultats. Tout ceci ajoute encore du traitement... accès à la queue des tâches, suppression des objets qu’elle contient… Au final, cela s’avère très pénalisant.

Il faut donc trouver un compromis : c'est là que la notion de seuil apparaît. Il faut qu’au dessous d'un certain niveau, le découpage en sous tâches n’ait plus lieu, évitant ainsi la multiplication des tâches à traiter.

Ajoutons à notre code les lignes ci-dessous et conservons notre méthode sequentialCompute() pour comparer le gain :

public class Fibonacci extends RecursiveAction {

      private final int number;

      private int result;

      private static final int THRESHOLD = 10;

      public Fibonacci(int number) {

            this.number = number;

      }

      public void compute() {

            int n = number;

            // seuil de granularité sous lequel on ne parallèlise plus

            if (n <= THRESHOLD) {

                  result = seqFib(n);

            } else {

                  final Fibonacci f1 = new Fibonacci(n - 1); // subdivision

                  final Fibonacci f2 = new Fibonacci(n - 2);

                  // invocation en parallèle des sous-taches (fork & join)

                  forkJoin(f1, f2);

                  // aggregation des resultats

                  result = f1.getResult() + f2.getResult();

            }

      }

      private int seqFib(int n) {

            if (n <= 1) {

                  return n;

            } else {

                  return seqFib(n - 1) + seqFib(n - 2);

            }

      }

      public void sequentialCompute() {

            result = seqFib(this.number);

      }

      public int getResult() {

            return result;

      }

      public static void main(String[] args) {

            final ForkJoinPool pool = new ForkJoinPool(2);

            Fibonacci f = new Fibonacci(44);

            pool.invoke(f);

            System.out.println(f.getResult());

      }

}

Ici, nous laissons le soin de choisir le seuil (en utilisant le constructeur comportant le paramètre « threshold ») ou bien de le laisser être évalué par l'autre constructeur (via une règle simple : seuil = p/2).

Il apparaît assez rapidement que cette valeur de seuil peut devenir critique car avoir une forte incidence sur les performances. Il appartient donc au développeur de trouver la bonne stratégie pour l'évaluer. Celle retenue ici n’est sans doute pas la meilleure mais pas la pire ;-)

Ces deux exemples permettent donc d’ouvrir l’esprit sur les développements possibles qu’il est permis de faire avec cet ensemble de classes de la JSR.

A coté de ces classes héritant de ForkJoinTask, la JSR166y propose quelques autres facilités, qui ne seront peut être pas incorporées dans Java 7 aux dernières nouvelles, mais qui seront accessibles via une librairie tierce si tel est le cas.

Le principe du ParallelArray est simple : proposer une structure pour manipuler un tableau de valeurs en s'appuyant de façon transparente sur les fonctionnalités de parallélisation pour le traitement de certaines tâches.

Exemple : déterminer la valeur maximale parmi les valeurs du tableau ou calculer la moyenne des valeurs du tableau...

L'exemple traité ci-après consiste en un service d'analyse de population d'individus. Les caractéristiques suivies vont être des indicateurs physiques : l'indice de masse corporelle et l'indice de masse grasse.

L'idée est de pouvoir identifier parmi une population de personnes, celles présentant un indice élevé ou faible (à risque dans les 2 cas), de subdiviser cette population selon des critères et pouvoir toujours travailler sur ces sous ensembles...

Un code valant mieux qu'un long discours, posons les bases de l'exemple :

    public enum Gender {

        MALE,

        FEMALE;

    }

    private double weight;

    private double height;

    private int age;

    private String name;

    private Person.Gender gender;

    public Person() {

    }

    public Person(final String name, final Gender gender, final int age, final double weight, final double height) {

        this.name = name;

        this.gender = gender;

        this.age = age;

        this.weight = weight;

        this.height = height;

    }

    [...]

    // Get/Setters...

    [...]

}

public class HealthAnalyzerService {

    private final ForkJoinPool pool = new ForkJoinPool(Runtime.getRuntime().availableProcessors());

    private Person[] people;

    private ParallelArray<Person> peopleArray;

    public void setPeople(final Person[] people) {

        this.people = people;

        this.peopleArray = ParallelArray.createFromCopy(people, pool);

    }

}

Cette fois-ci on entre dans le vif du sujet. Commençons les explications dans l'ordre de lecture :

Les attributs bmiOp et fmiOp sont les opérations (Ops.*) de base sur lesquelles le code du ParallelArray s'appuiera pour évaluer les 2 indices.

Pour une personne donnée, chaque opération évalue un indice selon divers paramètres.

NOTE : il existe un grand nombre d'opérations permettant d'effectuer un calcul sur un élément d'un ParallelArray.

La méthode getFilteredMappedArray(..) admet n paramètres de type Ops.Predicate. Ces paramètres vont servir à filtrer le tableau pour dégager une vue restreinte des données sur lesquelles travailler.

Les méthodes getLowest* / getHighest* qui suivent vont faire usage conjoint des éléments précités. Par exemple getHighestBMI(..) va d'abord extraire l'ensemble des données correspondant aux critères en entrée (les "Predicates") pour ensuite déterminer le max() résultant de l'opération bmiOp.

Si vous avez suivi ce cheminement, tout va être très clair maintenant pour la suite.

Les dernières méthodes de type getPersonWith* utilisent une autre fonctionnalité offerte par le ParallelArray : déterminer l'index d'un élément sur la base de critères.

Prenons getPersonWithHighestBMI(..), elle comporte un appel à la méthode getBMISummary, laquelle retourne un objet contenant divers indicateurs issus de calculs effectués conjointement avec un ensemble de "Predicates" et une opération.

Cet objet retourné permet d'exposer l'indice de l'élément du tableau initial ayant la valeur retournée par l'opération la plus élevée dans notre cas.

Cet indice va permettre de retrouver simplement l'objet en question dans le tableau initial (people).

Maintenant où se cache la parallélisation ? Et bien dans chaque méthode manipulant les données du tableau wrappé : la détermination d'un maximum, d'un minimum, d'un sous ensemble d'éléments répondant à des critères...

Ainsi de façon transparente, en ne coûtant que quelques lignes de code additionnelles, il est possible de traiter de façon optimisée (du point de vue temps d'exécution perçu) des ensembles volumineux de données.

Bien sûr l'ensemble des classes utilitaires fournies ici vont se prêter à la majorité des usages sans pouvoir couvrir la totalité. Dans ces cas particuliers, il faudra retourner à l'approche plus bas niveau proposée par l'ensemble des classes FortJoinTask.

Nous pouvons enfin conclure ce tutoriel sur la JSR166y. Les 2 aspects présentés devraient vous permettre de mieux cerner le potentiel offert. Le JDK 7 approchant, le périmètre des fonctionnalités intégrées à la future version de Java sera vite connu et une librairie plus stable comportant un backport pour Java 6 et les parties non intégrées devrait aussi être disponible. Il sera ainsi possible de profiter de ceci sans pour cela attendre la généralisation de Java 7 dans les environnements de production.