Quels sont les avantages d'utiliser le nouveau framework fork / join plutôt que de simplement diviser la grande tâche en N sous-tâches au début, les envoyer à un pool de threads mis en cache (à partir des exécuteurs ) et attendre que chaque tâche soit terminée? Je ne vois pas comment l'utilisation de l'abstraction fork / join simplifie le problème ou rend la solution plus efficace par rapport à ce que nous avons depuis des années.

Par exemple, l'algorithme de flou parallélisé dans l' exemple du didacticiel pourrait être implémenté comme ceci:

public class Blur implements Runnable {
    private int[] mSource;
    private int mStart;
    private int mLength;
    private int[] mDestination;

    private int mBlurWidth = 15; // Processing window size, should be odd.

    public ForkBlur(int[] src, int start, int length, int[] dst) {
        mSource = src;
        mStart = start;
        mLength = length;
        mDestination = dst;
    }

    public void run() {
        computeDirectly();
    }

    protected void computeDirectly() {
        // As in the example, omitted for brevity
    }
}

Divisez au début et envoyez les tâches à un pool de threads:

// source image pixels are in src
// destination image pixels are in dst
// threadPool is a (cached) thread pool

int maxSize = 100000; // analogous to F-J's "sThreshold"
List<Future> futures = new ArrayList<Future>();

// Send stuff to thread pool:
for (int i = 0; i < src.length; i+= maxSize) {
    int size = Math.min(maxSize, src.length - i);
    ForkBlur task = new ForkBlur(src, i, size, dst);
    Future f = threadPool.submit(task);
    futures.add(f);
}

// Wait for all sent tasks to complete:
for (Future future : futures) {
    future.get();
}

// Done!

Les tâches vont dans la file d'attente du pool de threads, à partir de laquelle elles sont exécutées lorsque les threads de travail deviennent disponibles. Tant que le fractionnement est suffisamment granulaire (pour éviter d'avoir à attendre particulièrement la dernière tâche) et que le pool de threads a suffisamment de threads (au moins N de processeurs), tous les processeurs fonctionnent à pleine vitesse jusqu'à ce que tout le calcul soit terminé.

Est-ce que je manque quelque chose? Quelle est la valeur ajoutée de l'utilisation du framework fork / join?

Je pense que le malentendu de base est que les exemples Fork / Join ne montrent PAS le vol de travail mais seulement une sorte de division et de conquête standard.

Le vol de travail serait comme ceci: le travailleur B a terminé son travail. Il est gentil, alors il regarde autour de lui et voit le travailleur A qui travaille toujours très dur. Il s'approche et demande: "Hé mon garçon, je pourrais te donner un coup de main." A répond. "Cool, j'ai cette tâche de 1000 unités. Jusqu'à présent, j'ai terminé 345 laissant 655. Pourriez-vous s'il vous plaît travailler sur les numéros 673 à 1000, je vais faire les 346 à 672." B dit "OK, commençons pour que nous puissions aller au pub plus tôt".

Vous voyez - les travailleurs doivent communiquer entre eux même lorsqu'ils ont commencé le vrai travail. C'est la partie manquante dans les exemples.

Les exemples en revanche ne montrent que quelque chose comme "utiliser des sous-traitants":

Ouvrier A: "Dang, j'ai 1000 unités de travail. Trop pour moi. Je vais en faire 500 moi-même et en sous-traiter 500 à quelqu'un d'autre." Cela continue jusqu'à ce que la grande tâche soit divisée en petits paquets de 10 unités chacun. Ceux-ci seront exécutés par les ouvriers disponibles. Mais si un paquet est une sorte de pilule empoisonnée et prend beaucoup plus de temps que les autres paquets - malchance, la phase de division est terminée.

La seule différence restante entre Fork / Join et fractionner la tâche à l'avance est la suivante: lors de la division initiale, la file d'attente de travail est pleine dès le début. Exemple: 1000 unités, le seuil est de 10, donc la file d'attente a 100 entrées. Ces paquets sont distribués aux membres du pool de threads.

Fork / Join est plus complexe et essaie de réduire le nombre de paquets dans la file d'attente:

• Étape 1: Mettez un paquet contenant (1 ... 1000) dans la file d'attente
• Étape 2: Un travailleur fait apparaître le paquet (1 ... 1000) et le remplace par deux paquets: (1 ... 500) et (501 ... 1000).
• Étape 3: Un ouvrier fait sauter le paquet (500 ... 1000) et pousse (500 ... 750) et (751 ... 1000).
• Etape n: La pile contient ces paquets: (1..500), (500 ... 750), (750 ... 875) ... (991..1000)
• Étape n + 1: le paquet (991..1000) est sauté et exécuté
• Étape n + 2: le paquet (981..990) est sauté et exécuté
• Étape n + 3: le paquet (961..980) est sauté et divisé en (961 ... 970) et (971..980). ....

Vous voyez: dans Fork / Join, la file d'attente est plus petite (6 dans l'exemple) et les phases "split" et "work" sont entrelacées.

Lorsque plusieurs travailleurs sautent et poussent simultanément, les interactions ne sont pas si claires, bien sûr.

Si vous avez n threads occupés qui fonctionnent tous à 100% indépendamment, ce sera mieux que n threads dans un pool Fork-Join (FJ). Mais cela ne fonctionne jamais de cette façon.

Il pourrait ne pas être en mesure de diviser précisément le problème en n parties égales. Même si vous le faites, la planification des threads n'est pas juste. Vous finirez par attendre le thread le plus lent. Si vous avez plusieurs tâches, elles peuvent chacune s'exécuter avec moins de parallélisme à n voies (généralement plus efficace), tout en passant à n voies lorsque les autres tâches sont terminées.

Alors pourquoi ne pas découper le problème en morceaux de taille FJ et faire travailler un pool de threads là-dessus. L'utilisation typique de FJ coupe le problème en petits morceaux. Les faire dans un ordre aléatoire nécessite beaucoup de coordination au niveau matériel. Les frais généraux seraient un tueur. Dans FJ, les tâches sont placées dans une file d'attente que le thread lit dans l'ordre Last In First Out (LIFO / pile), et le vol de travail (dans le travail de base, généralement) est effectué First In First Out (FIFO / "queue"). Le résultat est que le traitement de la longue matrice peut être effectué en grande partie de manière séquentielle, même s'il est divisé en petits morceaux. (Il est également vrai qu'il n'est peut-être pas anodin de diviser le problème en petits morceaux de taille égale en un seul big bang. Disons qu'il s'agit d'une certaine forme de hiérarchie sans équilibrage.)

Conclusion: FJ permet une utilisation plus efficace des threads matériels dans des situations inégales, ce qui sera toujours le cas si vous avez plus d'un thread.

Le but ultime des pools de threads et de Fork / Join est le même: les deux veulent utiliser au mieux la puissance CPU disponible pour un débit maximal. Le débit maximal signifie que le plus grand nombre de tâches possible doit être achevé sur une longue période. Que faut-il pour cela? (Pour ce qui suit, nous supposerons que les tâches de calcul ne manquent pas: il y a toujours assez à faire pour une utilisation à 100% du processeur. De plus, j'utilise de manière équivalente "CPU" pour les cœurs ou les cœurs virtuels en cas d'hyper-threading).

1. Au moins, il doit y avoir autant de threads en cours d'exécution que de processeurs disponibles, car exécuter moins de threads laissera un cœur inutilisé.
2. Au maximum, il doit y avoir autant de threads en cours d'exécution qu'il y a de processeurs disponibles, car l'exécution de plus de threads créera une charge supplémentaire pour le planificateur qui attribue des CPU aux différents threads, ce qui fait passer du temps CPU au planificateur plutôt qu'à notre tâche de calcul.

Ainsi, nous avons compris que pour un débit maximal, nous devons avoir exactement le même nombre de threads que les processeurs. Dans l'exemple de flou d'Oracle, vous pouvez à la fois prendre un pool de threads de taille fixe avec le nombre de threads égal au nombre de processeurs disponibles ou utiliser un pool de threads. Cela ne fera aucune différence, vous avez raison!

Alors, quand aurez-vous des problèmes avec un pool de threads? C'est si un thread se bloque , car votre thread attend qu'une autre tâche se termine. Supposons l'exemple suivant:

class AbcAlgorithm implements Runnable {
    public void run() {
        Future<StepAResult> aFuture = threadPool.submit(new ATask());
        StepBResult bResult = stepB();
        StepAResult aResult = aFuture.get();
        stepC(aResult, bResult);
    }
}

Ce que nous voyons ici est un algorithme qui se compose de trois étapes A, B et C.A et B peuvent être exécutées indépendamment l'une de l'autre, mais l'étape C a besoin du résultat des étapes A ET B.Ce que cet algorithme fait est de soumettre la tâche A à le threadpool et exécutez la tâche b directement. Après cela, le thread attendra que la tâche A soit également effectuée et continue avec l'étape C. Si A et B sont terminés en même temps, tout va bien. Mais que faire si A prend plus de temps que B? Cela peut être dû au fait que la nature de la tâche A l'exige, mais cela peut également être le cas car il n'y a pas de thread pour la tâche A disponible au début et la tâche A doit attendre. (S'il n'y a qu'un seul processeur disponible et que votre threadpool n'a donc qu'un seul thread, cela entraînera même un blocage, mais pour le moment, cela n'a pas d'importance). Le fait est que le thread qui vient d'exécuter la tâche Bbloque tout le fil . Comme nous avons le même nombre de threads que les processeurs et qu'un thread est bloqué, cela signifie qu'un processeur est inactif .

Fork / Join résout ce problème: dans le framework fork / join, vous écririez le même algorithme comme suit:

class AbcAlgorithm implements Runnable {
    public void run() {
        ATask aTask = new ATask());
        aTask.fork();
        StepBResult bResult = stepB();
        StepAResult aResult = aTask.join();
        stepC(aResult, bResult);
    }
}

Ça a l'air pareil, n'est-ce pas? Cependant, l'indice est que aTask.join cela ne bloquera pas . Au lieu de cela, c'est là que le vol de travail entre en jeu: le fil cherchera d'autres tâches qui ont été fourchues dans le passé et continuera avec celles-ci. Tout d'abord, il vérifie si les tâches qu'il a lui-même fourchues ont commencé à être traitées. Donc, si A n'a pas encore été démarré par un autre thread, il fera A ensuite, sinon il vérifiera la file d'attente des autres threads et leur volera leur travail. Une fois cette autre tâche d'un autre thread terminée, il vérifiera si A est terminé maintenant. Si c'est l'algorithme ci-dessus peut appeler stepC. Sinon, il cherchera encore une autre tâche à voler. Ainsi, les pools fork / join peuvent atteindre 100% d'utilisation du processeur, même face à des actions de blocage .

Cependant, il y a un piège: le vol de travail n'est possible que pour l' joinappel de l' ForkJoinTaskart. Cela ne peut pas être fait pour des actions de blocage externes telles que l'attente d'un autre thread ou l'attente d'une action d'E / S. Alors qu'en est-il de cela, attendre la fin des E / S est une tâche courante? Dans ce cas, si nous pouvions ajouter un thread supplémentaire au pool Fork / Join qui sera à nouveau arrêté dès que l'action de blocage sera terminée, sera la deuxième meilleure chose à faire. Et le ForkJoinPoolpeut réellement faire exactement cela si nous utilisons l' ManagedBlockerart.

Fibonacci

Dans le JavaDoc pour RecursiveTask est un exemple de calcul des nombres de Fibonacci à l'aide de Fork / Join. Pour une solution récursive classique, voir:

public static int fib(int n) {
    if (n <= 1) {
        return n;
    }
    return fib(n - 1) + fib(n - 2);
}

Comme cela est expliqué dans les JavaDocs, c'est une jolie façon de calculer les nombres de fibonacci, car cet algorithme a une complexité O (2 ^ n) alors que des moyens plus simples sont possibles. Cependant, cet algorithme est très simple et facile à comprendre, nous nous en tenons donc à lui. Supposons que nous voulions accélérer cela avec Fork / Join. Une implémentation naïve ressemblerait à ceci:

class Fibonacci extends RecursiveTask<Long> {
    private final long n;

    Fibonacci(long n) {
        this.n = n;
    }

    public Long compute() {
        if (n <= 1) {
            return n;
        }
        Fibonacci f1 = new Fibonacci(n - 1);
        f1.fork();
        Fibonacci f2 = new Fibonacci(n - 2);
        return f2.compute() + f1.join();
   }
}

Les étapes dans lesquelles cette tâche est divisée sont beaucoup trop courtes et cela fonctionnera donc horriblement, mais vous pouvez voir comment le cadre fonctionne généralement très bien: les deux sommets peuvent être calculés indépendamment, mais nous avons alors besoin des deux pour construire le final. résultat. Donc, la moitié est faite dans un autre fil. Amusez-vous à faire de même avec des pools de threads sans vous bloquer (possible, mais pas aussi simple).

Juste pour être complet: Si vous souhaitez réellement calculer les nombres de Fibonacci en utilisant cette approche récursive, voici une version optimisée:

class FibonacciBigSubtasks extends RecursiveTask<Long> {
    private final long n;

    FibonacciBigSubtasks(long n) {
        this.n = n;
    }

    public Long compute() {
        return fib(n);
    }

    private long fib(long n) {
        if (n <= 1) {
            return 1;
        }
        if (n > 10 && getSurplusQueuedTaskCount() < 2) {
            final FibonacciBigSubtasks f1 = new FibonacciBigSubtasks(n - 1);
            final FibonacciBigSubtasks f2 = new FibonacciBigSubtasks(n - 2);
            f1.fork();
            return f2.compute() + f1.join();
        } else {
            return fib(n - 1) + fib(n - 2);
        }
    }
}

Cela réduit considérablement les sous-tâches car elles ne sont fractionnées que lorsque n > 10 && getSurplusQueuedTaskCount() < 2est vrai, ce qui signifie qu'il y a beaucoup plus de 100 appels de méthode à faire ( n > 10) et qu'il n'y a pas de tâches man en attente ( getSurplusQueuedTaskCount() < 2).

Sur mon ordinateur (4 cœurs (8 en comptant Hyper-threading), Intel (R) Core (TM) i7-2720QM CPU @ 2,20 GHz), cela fib(50)prend 64 secondes avec l'approche classique et seulement 18 secondes avec l'approche Fork / Join qui est un gain tout à fait notable, mais pas autant que théoriquement possible.

Résumé

• Oui, dans votre exemple, Fork / Join n'a aucun avantage sur les pools de threads classiques.
• Fork / Join peut considérablement améliorer les performances en cas de blocage
• Fork / Join évite certains problèmes de blocage

Fork / join est différent d'un pool de threads car il implémente le vol de travail. Depuis Fork / Join

Comme avec n'importe quel ExecutorService, le framework fork / join distribue les tâches aux threads de travail dans un pool de threads. Le framework fork / join est distinct car il utilise un algorithme de vol de travail. Les threads de travail qui n'ont plus rien à faire peuvent voler des tâches à d'autres threads qui sont encore occupés.

Supposons que vous ayez deux threads et 4 tâches a, b, c, d qui prennent respectivement 1, 1, 5 et 6 secondes. Initialement, a et b sont affectés au thread 1 et c et d au thread 2. Dans un pool de threads, cela prendrait 11 secondes. Avec fork / join, le thread 1 se termine et peut voler le travail du thread 2, donc la tâche d finirait par être exécutée par le thread 1. Thread 1 exécute a, b et d, thread 2 juste c. Temps total: 8 secondes, pas 11.

EDIT: Comme le souligne Joonas, les tâches ne sont pas nécessairement pré-allouées à un thread. L'idée de fork / join est qu'un thread peut choisir de diviser une tâche en plusieurs sous-éléments. Donc, pour reformuler ce qui précède:

Nous avons deux tâches (ab) et (cd) qui prennent respectivement 2 et 11 secondes. Le thread 1 commence à exécuter ab et le divise en deux sous-tâches a et b. De même avec le thread 2, il se divise en deux sous-tâches c & d. Lorsque le fil 1 a terminé a et b, il peut voler d au fil 2.

Tout le monde ci-dessus a raison, les avantages sont obtenus par le vol de travail, mais pour expliquer pourquoi.

Le principal avantage est la coordination efficace entre les threads de travail. Le travail doit être divisé et réassemblé, ce qui nécessite une coordination. Comme vous pouvez le voir dans la réponse d'AH ci-dessus, chaque fil a sa propre liste de travail. Une propriété importante de cette liste est qu'elle est triée (grandes tâches en haut et petites tâches en bas). Chaque thread exécute les tâches en bas de sa liste et vole les tâches du haut des autres listes de threads.

Le résultat est:

• La tête et la queue des listes de tâches peuvent être synchronisées indépendamment, ce qui réduit les conflits sur la liste.
• Les sous-arbres importants du travail sont divisés et réassemblés par le même thread, donc aucune coordination inter thread n'est requise pour ces sous-arbres.
• Lorsqu'un fil vole du travail, il prend un gros morceau qu'il subdivise ensuite dans sa propre liste
• Le travail d'acier signifie que les filets sont presque entièrement utilisés jusqu'à la fin du processus.

La plupart des autres schémas de division et de conquête utilisant des pools de threads nécessitent davantage de communication et de coordination entre les threads.

Dans cet exemple, Fork / Join n'ajoute aucune valeur car la fourche n'est pas nécessaire et la charge de travail est répartie uniformément entre les threads de travail. Fork / Join n'ajoute que des frais généraux.

Voici un bel article sur le sujet. Citation:

Dans l'ensemble, nous pouvons dire que ThreadPoolExecutor doit être préféré lorsque la charge de travail est uniformément répartie entre les threads de travail. Pour pouvoir garantir cela, vous devez savoir précisément à quoi ressemblent les données d'entrée. En revanche, le ForkJoinPool offre de bonnes performances quelles que soient les données d'entrée et constitue donc une solution nettement plus robuste.

Une autre différence importante semble être qu'avec FJ, vous pouvez effectuer plusieurs phases complexes de «jointure». Considérez le tri de fusion de http://faculty.ycp.edu/~dhovemey/spring2011/cs365/lecture/lecture18.html , il y aurait trop d'orchestration nécessaire pour pré-fractionner ce travail. Par exemple, vous devez faire les choses suivantes:

• trier le premier trimestre
• trier le deuxième trimestre
• fusionner les 2 premiers trimestres
• trier le troisième trimestre
• trier le quatrième trimestre
• fusionner les 2 derniers trimestres
• fusionner les 2 moitiés

Comment spécifiez-vous que vous devez faire les tris avant les fusions qui les concernent etc.

J'ai cherché la meilleure façon de faire une certaine chose pour chacun d'une liste d'articles. Je pense que je vais juste pré-diviser la liste et utiliser un ThreadPool standard. FJ semble plus utile lorsque le travail ne peut pas être pré-divisé en suffisamment de tâches indépendantes, mais peut être divisé récursivement en tâches indépendantes entre elles (par exemple, le tri des moitiés est indépendant mais la fusion des 2 moitiés triées en un tout trié ne l'est pas).

F / J a également un avantage distinct lorsque vous avez des opérations de fusion coûteuses. Comme il se divise en une structure arborescente, vous ne faites que des fusions log2 (n) au lieu de n fusions avec la division linéaire des threads. (Cela fait l'hypothèse théorique que vous avez autant de processeurs que de threads, mais c'est toujours un avantage) Pour un travail à domicile, nous avons dû fusionner plusieurs milliers de tableaux 2D (toutes les mêmes dimensions) en additionnant les valeurs à chaque index. Avec les processeurs fork join et P, le temps approche log2 (n) lorsque P approche l'infini.

1 2 3 .. 7 3 1 .... 8 5 4
4 5 6 + 2 4 3 => 6 9 9
7 8 9 .. 1 1 0 .... 8 9 9

Vous seriez surpris des performances de ForkJoin dans des applications telles que le robot d'exploration. voici le meilleur tutoriel dont vous pourriez tirer des leçons.

La logique de Fork / Join est très simple: (1) séparer (fork) chaque grande tâche en tâches plus petites; (2) traiter chaque tâche dans un thread distinct (en les séparant en tâches encore plus petites si nécessaire); (3) rejoindre les résultats.

Si le problème est tel que nous devons attendre la fin des autres threads (comme dans le cas du tri du tableau ou de la somme du tableau), une jointure de fourche doit être utilisée, car Executor (Executors.newFixedThreadPool (2)) s'étouffera en raison de le nombre de fils. Le pool forkjoin créera plus de threads dans ce cas pour couvrir le thread bloqué pour maintenir le même parallélisme

Source: http://www.oracle.com/technetwork/articles/java/fork-join-422606.html

Le problème avec les exécuteurs pour implémenter les algorithmes de division et de conquête n'est pas lié à la création de sous-tâches, car un appelable est libre de soumettre une nouvelle sous-tâche à son exécuteur et d'attendre son résultat de manière synchrone ou asynchrone. Le problème est celui du parallélisme: lorsqu'un Callable attend le résultat d'un autre Callable, il est mis dans un état d'attente, perdant ainsi l'occasion de gérer un autre Callable mis en file d'attente pour l'exécution.

Le framework fork / join ajouté au package java.util.concurrent dans Java SE 7 grâce aux efforts de Doug Lea comble cette lacune

Source: https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/ForkJoinPool.html

Le pool tente de maintenir suffisamment de threads actifs (ou disponibles) en ajoutant, suspendant ou reprenant dynamiquement des threads de travail internes, même si certaines tâches sont bloquées en attendant d'en rejoindre d'autres. Cependant, aucun ajustement de ce type n'est garanti face à des E / S bloquées ou à toute autre synchronisation non gérée.

public int getPoolSize () Renvoie le nombre de threads de travail qui ont démarré mais pas encore terminés. Le résultat renvoyé par cette méthode peut différer de getParallelism () lorsque des threads sont créés pour maintenir le parallélisme lorsque d'autres sont bloqués en coopération.

Je voudrais ajouter une réponse courte pour ceux qui n'ont pas beaucoup de temps pour lire de longues réponses. La comparaison est tirée du livre Applied Akka Patterns:

Votre décision d'utiliser un exécuteur fork-join-executor ou un thread-pool-executor dépend en grande partie du fait que les opérations dans ce répartiteur seront bloquantes. Un exécuteur fork-join-vous donne un nombre maximum de threads actifs, alors qu'un thread-pool-executor vous donne un nombre fixe de threads. Si les threads sont bloqués, un exécuteur fork-join-executor en créera plus, contrairement à un thread-pool-executor. Pour les opérations de blocage, vous êtes généralement mieux avec un exécuteur de pool de threads car il empêche votre nombre de threads d'exploser. Des opérations plus «réactives» sont meilleures dans un exécuteur fork-join-executor.