Java 7, divide y vencerás.

El siguiente ejemplo ejecuta el rellenado de un array de bytes de un tamaño importante de dos maneras:

1. La primera de ellas, de manera secuencial. Recorre todas las posiciones, y rellena cada una de ellas con un número aleatorio
2. En la segunda de ellas, realiza el mismo algoritmo en paralelo. Divide el bloque grande en bloques más pequeños, y cada uno de ellos forma parte de una tarea global (comparten memoria, cada uno de ellos tiene conocimientos de su tarea específica...). Se aprovecha entonces, en el caso del que escribe, los dos procesadores del núcleo de la CPU.

Muy interesante este concepto, a tener en cuenta para tareas grandes susceptibles de ser divididas en tareas más pequeñas.

Me recordó un poco al principio inicial de MapReduce (divide una tarea grande en tareas pequeñas.... y vencerás.

La salida, en el caso del que escribe, fue la siguiente:

Sequential processing time: 1683 ms

Sequential processing time: 1662 ms

Number of processor available: 2

Parallel processing time: 930 ms

Parallel processing time: 851 ms

Number of steals: 598

Saludos cordiales y gracias por su atención, como siempre un gusto

package com.sprunck.sample;

import static java.util.Arrays.asList;

import java.util.Random;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;

public class ForkJoinRandomFillAction {
	Random random = new Random();

	public void loadArray(int[] array) {
		for (int i = 0; i < array.length; i++) {
			array[i] = random.nextInt(10000); // Generates numbers from 0 to
												// 10000
		}
	}

	public static void main(String[] args) {
		ForkJoinRandomFillAction sort = new ForkJoinRandomFillAction();

		int arrayLength = 1_00_00_0000;
		int array[] = new int[arrayLength];

		// No. of times sequential & Parallel operation should be performed to
		// warm up HotSpot JVM
		final int iterations = 2;

		for (int i = 0; i < iterations; i++) {
			long start = System.currentTimeMillis();
			sort.loadArray(array);

			System.out.println("Sequential processing time: "
					+ (System.currentTimeMillis() - start) + " ms");

		}

		System.out.println("Number of processor available: "
				+ Runtime.getRuntime().availableProcessors());

		ForkJoinPool fjpool = new ForkJoinPool(64); // Default parallelism level
													// =
													// Runtime.getRuntime().availableProcessors()

		for (int i = 0; i < iterations; i++) {
			// Create a task with the complete array
			RecursiveAction task = new RandomFillAction(array, 0, array.length);
			long start = System.currentTimeMillis();
			fjpool.invoke(task);

			System.out.println("Parallel processing time: "
					+ (System.currentTimeMillis() - start) + " ms");
		}

		System.out
				.println("Number of steals: " + fjpool.getStealCount() + "\n");
	}
}

class RandomFillAction extends RecursiveAction {
	private static final long serialVersionUID = 1L;
	final int low;
	final int high;
	private int[] array;
	final int splitSize = 40000; // Some threshold size to spit the task

	public RandomFillAction(int[] array, int low, int high) {
		this.low = low;
		this.high = high;
		this.array = array;
	}

	@Override
	protected void compute() {
		if (high - low > splitSize) {
			// task is huge so divide in half
			int mid = (low + high) / 2;
			invokeAll(asList(new RandomFillAction(array, low, mid),
					new RandomFillAction(array, mid, high)));
		} else {
			// Some calculation logic
			Random random = new Random();
			for (int i = low; i < high; i++) {
				array[i] = random.nextInt(10000);
			}
		}
	}
}

Comments

[antuansoft]

Muy interesante, la verdad que son capacidades que podemos usar en el día a día de la programación pero las ignoramos.

Estas librerías de java.util.concurrent.ForkJoinPool y RecursiveAction sólo están en Java 7 ¿verdad?

Muy útil!!!