¿Cuáles son los beneficios de usar el nuevo marco fork / join simplemente dividiendo la gran tarea en N subtareas al principio, enviándolas a un grupo de subprocesos en caché (de Ejecutores ) y esperando que se complete cada tarea? No veo cómo el uso de la abstracción fork / join simplifica el problema o hace que la solución sea más eficiente de lo que hemos tenido durante años.

Por ejemplo, el algoritmo de desenfoque paralelo en el ejemplo del tutorial podría implementarse de esta manera:

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
    }
}

Partir al principio y enviar tareas a un grupo de subprocesos:

// 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!

Las tareas van a la cola del grupo de subprocesos, desde donde se ejecutan a medida que los subprocesos de trabajo están disponibles. Siempre y cuando la división sea lo suficientemente granular (para evitar tener que esperar particularmente la última tarea) y el grupo de subprocesos tenga suficientes subprocesos (al menos N de procesadores), todos los procesadores estarán trabajando a toda velocidad hasta que se complete todo el cálculo.

¿Me estoy perdiendo de algo? ¿Cuál es el valor agregado de usar el framework fork / join?

Creo que el malentendido básico es que los ejemplos de Fork / Join NO muestran robo de trabajo sino solo algún tipo de división y conquista estándar.

El robo de trabajo sería así: el trabajador B ha terminado su trabajo. Él es amable, así que mira a su alrededor y ve al Trabajador A todavía trabajando muy duro. Se acerca y pregunta: "Hola muchacho, podría echarte una mano". A responde. "Genial, tengo esta tarea de 1000 unidades. Hasta ahora he terminado 345 dejando 655. ¿Podría por favor trabajar en el número 673 a 1000, voy a hacer el 346 a 672". B dice "OK, comencemos para que podamos ir al pub antes".

Verá, los trabajadores deben comunicarse entre sí incluso cuando comenzaron el trabajo real. Esta es la parte que falta en los ejemplos.

Los ejemplos, por otro lado, muestran solo algo como "usar subcontratistas":

Trabajador A: "Dang, tengo 1000 unidades de trabajo. Demasiado para mí. Haré 500 yo mismo y subcontrataré 500 a otra persona". Esto continúa hasta que la gran tarea se divide en pequeños paquetes de 10 unidades cada uno. Estos serán ejecutados por los trabajadores disponibles. Pero si un paquete es una especie de píldora venenosa y toma mucho más tiempo que otros paquetes, mala suerte, la fase de división ha terminado.

La única diferencia que queda entre Fork / Join y dividir la tarea por adelantado es esta: al dividir por adelantado, tiene la cola de trabajo llena desde el principio. Ejemplo: 1000 unidades, el umbral es 10, por lo que la cola tiene 100 entradas. Estos paquetes se distribuyen a los miembros del conjunto de hilos.

Fork / Join es más complejo e intenta mantener el número de paquetes en la cola más pequeño:

• Paso 1: coloque un paquete que contenga (1 ... 1000) en la cola
• Paso 2: un trabajador saca el paquete (1 ... 1000) y lo reemplaza con dos paquetes: (1 ... 500) y (501 ... 1000).
• Paso 3: un trabajador abre el paquete (500 ... 1000) y empuja (500 ... 750) y (751 ... 1000).
• Paso n: La pila contiene estos paquetes: (1..500), (500 ... 750), (750 ... 875) ... (991..1000)
• Paso n + 1: el paquete (991..1000) aparece y se ejecuta
• Paso n + 2: el paquete (981..990) aparece y se ejecuta
• Paso n + 3: el paquete (961..980) aparece y se divide en (961 ... 970) y (971..980). ....

Verá: en Fork / Join la cola es más pequeña (6 en el ejemplo) y las fases de "división" y "trabajo" están entrelazadas.

Cuando varios trabajadores aparecen y empujan simultáneamente, las interacciones no son tan claras, por supuesto.

Si tiene n subprocesos ocupados trabajando todos al 100% de forma independiente, será mejor que n subprocesos en un grupo de Fork-Join (FJ). Pero nunca funciona de esa manera.

Es posible que no pueda dividir el problema con precisión en n partes iguales. Incluso si lo hace, la programación de hilos está lejos de ser justa. Terminarás esperando el hilo más lento. Si tiene varias tareas, cada una de ellas puede ejecutarse con un paralelismo menor que el n-way (generalmente más eficiente), sin embargo, puede ir al n-way cuando otras tareas hayan finalizado.

Entonces, ¿por qué no simplemente cortamos el problema en pedazos de tamaño FJ y hacemos que un grupo de subprocesos funcione en eso? El uso típico de FJ corta el problema en pedazos pequeños. Hacer esto en un orden aleatorio requiere mucha coordinación a nivel de hardware. Los gastos generales serían un asesino. En FJ, las tareas se colocan en una cola que el subproceso lee en el orden Último en entrar, primero en salir (LIFO / stack), y el robo de trabajo (en el trabajo central, generalmente) se realiza en Primero en entrar, primero en salir (FIFO / "cola"). El resultado es que el procesamiento de matriz larga se puede realizar en gran medida de forma secuencial, a pesar de que se divide en pequeños fragmentos. (También es el caso de que podría no ser trivial dividir el problema en pequeños trozos de tamaño uniforme en una gran explosión. Digamos que lidiar con alguna forma de jerarquía sin equilibrar).

Conclusión: FJ permite un uso más eficiente de los hilos de hardware en situaciones desiguales, lo cual será siempre si tiene más de un hilo.

El objetivo final de los grupos de subprocesos y Fork / Join es similar: ambos quieren utilizar la potencia de CPU disponible lo mejor que puedan para obtener el máximo rendimiento. El rendimiento máximo significa que se deben completar tantas tareas como sea posible en un largo período de tiempo. ¿Qué se necesita para hacer eso? (Para lo siguiente asumiremos que no faltan las tareas de cálculo: siempre hay suficiente para hacer una utilización del 100% de la CPU. Además, uso "CPU" de manera equivalente para núcleos o núcleos virtuales en caso de hiperprocesamiento).

1. Al menos debe haber tantos subprocesos en ejecución como CPU disponibles, porque ejecutar menos subprocesos dejará un núcleo sin usar.
2. Como máximo, debe haber tantos subprocesos en ejecución como CPU disponibles, porque la ejecución de más subprocesos creará una carga adicional para el Programador que asigna CPU a los diferentes subprocesos, lo que hace que algo de tiempo de CPU vaya al programador en lugar de nuestra tarea computacional.

Por lo tanto, descubrimos que para obtener el máximo rendimiento necesitamos tener exactamente el mismo número de subprocesos que las CPU. En el ejemplo borroso de Oracle, ambos pueden tomar un grupo de subprocesos de tamaño fijo con el número de subprocesos igual al número de CPU disponibles o utilizar un grupo de subprocesos. No hará la diferencia, tienes razón!

Entonces, ¿cuándo te meterás en problemas con un grupo de subprocesos? Eso es si un hilo se bloquea , porque su hilo está esperando que se complete otra tarea. Supongamos el siguiente ejemplo:

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

Lo que vemos aquí es un algoritmo que consta de tres pasos A, B y C. A y B pueden realizarse independientemente uno del otro, pero el paso C necesita el resultado de los pasos A y B. Lo que hace este algoritmo es enviar la tarea A a el conjunto de subprocesos y realizar la tarea b directamente. Después de eso, el subproceso esperará a que se realice la tarea A también y continuará con el paso C. Si A y B se completan al mismo tiempo, entonces todo está bien. Pero, ¿qué pasa si A tarda más que B? Esto puede deberse a que la naturaleza de la tarea A lo dicta, pero también puede ser el caso porque no hay un hilo para la tarea A disponible al principio y la tarea A debe esperar. (Si solo hay una única CPU disponible y, por lo tanto, su grupo de subprocesos tiene solo un solo subproceso, esto incluso provocará un punto muerto, pero por ahora eso está fuera del punto). El punto es que el hilo que acaba de ejecutar la tarea Bbloquea todo el hilo . Como tenemos el mismo número de subprocesos que las CPU y un subproceso está bloqueado, eso significa que una CPU está inactiva .

Fork / Join resuelve este problema: en el framework fork / join escribirías el mismo algoritmo de la siguiente manera:

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

Se ve igual, ¿no? Sin embargo, la pista es que aTask.join no se bloqueará . En cambio, aquí es donde entra en juego el robo de trabajo : el hilo buscará otras tareas que se bifurcaron en el pasado y continuará con ellas. Primero verifica si las tareas que se bifurcó se han comenzado a procesar. Entonces, si A no ha sido iniciado por otro hilo todavía, hará A a continuación, de lo contrario verificará la cola de otros hilos y robará su trabajo. Una vez que se haya completado esta otra tarea de otro subproceso, comprobará si A se ha completado ahora. Si es el algoritmo anterior puede llamar stepC. De lo contrario, buscará otra tarea más para robar. Por lo tanto, los grupos fork / join pueden lograr un 100% de utilización de la CPU, incluso frente a acciones de bloqueo .

Sin embargo, hay una trampa: el robo de trabajo solo es posible para la joinllamada de ForkJoinTasks. No se puede hacer para acciones de bloqueo externo como esperar otro subproceso o esperar una acción de E / S. Entonces, ¿qué pasa con eso, esperar a que se complete la E / S es una tarea común? En este caso, si pudiéramos agregar un hilo adicional al grupo Fork / Join que se detendrá nuevamente tan pronto como se complete la acción de bloqueo, será la segunda mejor opción. Y en ForkJoinPoolrealidad puede hacer eso si estamos usando ManagedBlockers.

Fibonacci

En JavaDoc para RecursiveTask hay un ejemplo para calcular números de Fibonacci usando Fork / Join. Para una solución recursiva clásica ver:

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

Como se explica en JavaDocs, esta es una forma bastante simple de calcular los números de Fibonacci, ya que este algoritmo tiene una complejidad O (2 ^ n), mientras que las formas más simples son posibles. Sin embargo, este algoritmo es muy simple y fácil de entender, por lo que nos atenemos a él. Supongamos que queremos acelerar esto con Fork / Join. Una implementación ingenua se vería así:

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();
   }
}

Los pasos en los que se divide esta Tarea son demasiado cortos y, por lo tanto, funcionarán horriblemente, pero puede ver cómo el marco generalmente funciona muy bien: los dos sumandos se pueden calcular de forma independiente, pero luego los necesitamos para construir el final resultado. Entonces la mitad se hace en otro hilo. Diviértete haciendo lo mismo con los grupos de subprocesos sin llegar a un punto muerto (posible, pero no tan simple).

Solo para completar: si realmente desea calcular los números de Fibonacci utilizando este enfoque recursivo, aquí hay una versión optimizada:

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);
        }
    }
}

Esto mantiene las subtareas mucho más pequeñas porque solo se dividen cuando n > 10 && getSurplusQueuedTaskCount() < 2es verdadero, lo que significa que hay significativamente más de 100 llamadas a métodos para hacer ( n > 10) y no hay muchas tareas de hombre esperando ( getSurplusQueuedTaskCount() < 2).

En mi computadora (4 núcleos (8 cuando se cuenta Hyper-threading), Intel (R) Core (TM) i7-2720QM CPU @ 2.20GHz) fib(50)toma 64 segundos con el enfoque clásico y solo 18 segundos con el enfoque Fork / Join que es una ganancia bastante notable, aunque no tanto como teóricamente posible.

Resumen

• Sí, en su ejemplo, Fork / Join no tiene ninguna ventaja sobre los grupos de subprocesos clásicos.
• Fork / Join puede mejorar drásticamente el rendimiento cuando el bloqueo está involucrado
• Fork / Join evita algunos problemas de punto muerto

Fork / join es diferente de un grupo de subprocesos porque implementa el robo de trabajo. Desde Fork / Join

Al igual que con cualquier ExecutorService, el framework fork / join distribuye tareas a los subprocesos de trabajo en un grupo de subprocesos. El framework fork / join es distinto porque utiliza un algoritmo de robo de trabajo. Los hilos de trabajo que se quedan sin cosas que hacer pueden robar tareas de otros hilos que todavía están ocupados.

Digamos que tiene dos subprocesos y 4 tareas a, b, c, d que toman 1, 1, 5 y 6 segundos respectivamente. Inicialmente, ayb se asignan al subproceso 1 yc y d al subproceso 2. En un grupo de subprocesos, esto tomaría 11 segundos. Con fork / join, el subproceso 1 finaliza y puede robar el trabajo del subproceso 2, por lo que la tarea d terminaría siendo ejecutada por el subproceso 1. El subproceso 1 ejecuta a, byd, el subproceso 2 solo c. Tiempo total: 8 segundos, no 11.

EDITAR: como señala Joonas, las tareas no están necesariamente asignadas previamente a un hilo. La idea de fork / join es que un hilo puede elegir dividir una tarea en múltiples sub-piezas. Entonces, para reafirmar lo anterior:

Tenemos dos tareas (ab) y (cd) que toman 2 y 11 segundos respectivamente. El subproceso 1 comienza a ejecutar ab y lo divide en dos subtareas a y b. De manera similar con el hilo 2, se divide en dos subtareas c & d. Cuando el hilo 1 ha terminado a & b, puede robar d del hilo 2.

Todos los anteriores son correctos, los beneficios se logran con el robo de trabajo, pero para ampliar por qué es así.

El beneficio principal es la coordinación eficiente entre hilos de trabajo. El trabajo debe dividirse y volverse a montar, lo que requiere coordinación. Como puede ver en la respuesta de AH arriba, cada hilo tiene su propia lista de trabajo. Una propiedad importante de esta lista es que está ordenada (tareas grandes en la parte superior y tareas pequeñas en la parte inferior). Cada hilo ejecuta las tareas en la parte inferior de su lista y roba tareas de la parte superior de otras listas de hilos.

El resultado de esto es:

• El encabezado y la cola de las listas de tareas pueden sincronizarse independientemente, lo que reduce la contención en la lista.
• Los subárboles significativos del trabajo se dividen y se vuelven a ensamblar por el mismo subproceso, por lo que no se requiere coordinación entre subprocesos para estos subárboles.
• Cuando un hilo roba trabajo, toma una pieza grande que luego se subdivide en su propia lista
• El trabajo de acero significa que los hilos se utilizan casi por completo hasta el final del proceso.

La mayoría de los otros esquemas de divide y vencerás que usan grupos de subprocesos requieren más comunicación y coordinación entre subprocesos.

En este ejemplo, Fork / Join no agrega valor porque la bifurcación no es necesaria y la carga de trabajo se divide de manera uniforme entre los hilos de los trabajadores. Fork / Join solo agrega gastos generales.

Aquí hay un buen artículo sobre el tema. Citar:

En general, podemos decir que el ThreadPoolExecutor es preferible donde la carga de trabajo se divide de manera uniforme en los subprocesos de los trabajadores. Para poder garantizar esto, necesita saber con precisión cómo se ven los datos de entrada. Por el contrario, el ForkJoinPool proporciona un buen rendimiento independientemente de los datos de entrada y, por lo tanto, es una solución significativamente más robusta.

Otra diferencia importante parece ser que con FJ, puede hacer múltiples y complejas fases de "Unirse". Considere el tipo de fusión de http://faculty.ycp.edu/~dhovemey/spring2011/cs365/lecture/lecture18.html , se necesitaría demasiada orquestación para dividir previamente este trabajo. Por ejemplo, debe hacer lo siguiente:

• ordenar el primer trimestre
• ordenar el segundo trimestre
• fusionar los primeros 2 trimestres
• ordenar el tercer trimestre
• ordenar el cuarto trimestre
• fusionar los últimos 2 trimestres
• fusionar las 2 mitades

¿Cómo especifica que debe hacer los géneros antes de las fusiones que les conciernen, etc.

He estado buscando la mejor manera de hacer una determinada cosa para cada una de una lista de elementos. Creo que simplemente dividiré previamente la lista y usaré un ThreadPool estándar. FJ parece más útil cuando el trabajo no se puede dividir previamente en suficientes tareas independientes, pero se puede dividir recursivamente en tareas que son independientes entre sí (por ejemplo, ordenar las mitades son independientes pero fusionar las 2 mitades ordenadas en un todo ordenado no lo es).

F / J también tiene una clara ventaja cuando tiene costosas operaciones de fusión. Debido a que se divide en una estructura de árbol, solo se fusionan log2 (n) en lugar de n fusiones con división lineal de hilos. (Esto asume la suposición teórica de que tiene tantos procesadores como hilos, pero sigue siendo una ventaja) Para una tarea, tuvimos que fusionar varios miles de matrices 2D (todas las mismas dimensiones) sumando los valores en cada índice. Con la unión fork y los procesadores P, el tiempo se acerca a log2 (n) a medida que P se acerca al infinito.

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

Te sorprendería el rendimiento de ForkJoin en aplicaciones como el rastreador. Aquí está el mejor tutorial del que aprenderías.

La lógica de Fork / Join es muy simple: (1) separar (fork) cada tarea grande en tareas más pequeñas; (2) procesar cada tarea en un hilo separado (separándolas en tareas aún más pequeñas si es necesario); (3) unirse a los resultados.

Si el problema es tal que tenemos que esperar a que se completen otros subprocesos (como en el caso de la clasificación de la matriz o la suma de la matriz), se debe utilizar la unión de la bifurcación, ya que el Ejecutor (Executors.newFixedThreadPool (2)) se ahogará debido a la limitación Número de hilos. El grupo forkjoin creará más hilos en este caso para cubrir el hilo bloqueado para mantener el mismo paralelismo

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

El problema con los ejecutores para implementar algoritmos de divide y vencerás no está relacionado con la creación de subtareas, porque un invocable es libre de enviar una nueva subtarea a su ejecutor y esperar su resultado de forma síncrona o asíncrona. El problema es el paralelismo: cuando un invocable espera el resultado de otro invocable, se pone en estado de espera, desperdiciando así la oportunidad de manejar otro invocable en cola para su ejecución.

El marco fork / join agregado al paquete java.util.concurrent en Java SE 7 a través de los esfuerzos de Doug Lea llena ese vacío

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

El grupo intenta mantener suficientes subprocesos activos (o disponibles) agregando, suspendiendo o reanudando dinámicamente los subprocesos de trabajo internos, incluso si algunas tareas están detenidas esperando unirse a otras. Sin embargo, dichos ajustes no están garantizados frente a una E / S bloqueada u otra sincronización no administrada

public int getPoolSize () Devuelve el número de subprocesos de trabajo que se han iniciado pero aún no se han terminado. El resultado devuelto por este método puede diferir de getParallelism () cuando se crean subprocesos para mantener el paralelismo cuando otros se bloquean cooperativamente.

Me gustaría agregar una respuesta corta para aquellos que no tienen mucho tiempo para leer respuestas largas. La comparación está tomada del libro Patrones de Akka aplicados:

Su decisión sobre si utilizar un ejecutor de unión de horquilla o un ejecutor de grupo de subprocesos se basa en gran medida en si las operaciones en ese despachador se bloquearán. Un ejecutor de unión de fork le proporciona un número máximo de subprocesos activos, mientras que un ejecutor de grupo de subprocesos le proporciona un número fijo de subprocesos. Si los subprocesos están bloqueados, un ejecutor de unión de bifurcación creará más, mientras que un ejecutor de agrupación de subprocesos no. Para las operaciones de bloqueo, generalmente está mejor con un ejecutor de grupo de subprocesos porque evita que su número de subprocesos explote. Más operaciones "reactivas" son mejores en un fork-join-ejecutor.