Java: Paralelización ordenación rápida a través de multi-threading

Question

Estoy experimentando con paralelización de algoritmos en Java. Empecé con el ordenamiento por mezcla, y publicado mi intento en esta pregunta . Mi intento es revisada en el código siguiente, en la que ahora trato de paralelizar ordenación rápida.

¿Hay errores de novato en mi aplicación multi-roscado o acercamiento a este problema? Si no es así, ¿no debería esperar más que un aumento de velocidad del 32% entre una secuencia y un algoritmo parallelized en un duelo-core (ver horarios en la parte inferior)?

Aquí está el algoritmo multihilo:

    public class ThreadedQuick extends Thread
    {
        final int MAX_THREADS = Runtime.getRuntime().availableProcessors();

        CountDownLatch doneSignal;
        static int num_threads = 1;

        int[] my_array;
        int start, end;

        public ThreadedQuick(CountDownLatch doneSignal, int[] array, int start, int end) {
            this.my_array = array;
            this.start = start;
            this.end = end;
            this.doneSignal = doneSignal;
        }

        public static void reset() {
            num_threads = 1;
        }

        public void run() {
            quicksort(my_array, start, end);
            doneSignal.countDown();
            num_threads--;
        }

        public void quicksort(int[] array, int start, int end) {
            int len = end-start+1;

            if (len <= 1)
                return;

            int pivot_index = medianOfThree(array, start, end);
            int pivotValue = array[pivot_index];

            swap(array, pivot_index, end);

            int storeIndex = start;
            for (int i = start; i < end; i++) {
               if (array[i] <= pivotValue) {
                   swap(array, i, storeIndex);
                   storeIndex++;
               }
            }

            swap(array, storeIndex, end);

            if (num_threads < MAX_THREADS) {
                num_threads++;

                CountDownLatch completionSignal = new CountDownLatch(1);

                new ThreadedQuick(completionSignal, array, start, storeIndex - 1).start();
                quicksort(array, storeIndex + 1, end);

                try {
                    completionSignal.await(1000, TimeUnit.SECONDS);
                } catch(Exception ex) {
                    ex.printStackTrace();
                }
            } else {
                quicksort(array, start, storeIndex - 1);
                quicksort(array, storeIndex + 1, end);
            }
        }
    }

Aquí es cómo lo comienzo apagado:

ThreadedQuick.reset();
CountDownLatch completionSignal = new CountDownLatch(1);
new ThreadedQuick(completionSignal, array, 0, array.length-1).start();
try {
    completionSignal.await(1000, TimeUnit.SECONDS);
} catch(Exception ex){
    ex.printStackTrace();
}

He probado esto en contra Arrays.sort y un algoritmo de ordenación rápida secuencial similar. Estos son los resultados de temporización en un dell portátil Intel duelo núcleos, en segundo:

Elementos: 500.000, secuencial: 0.068592, roscada: 0.046871, Arrays.sort: 0.079677

Elementos: 1.000.000, secuencial: 0.14416, roscada: 0.095492, Arrays.sort: 0.167155

Elementos: 2.000.000, secuencial: 0.301666, roscada: 0.205719, Arrays.sort: 0.350982

Elementos: 4.000.000, secuencial: 0.623291, roscada: 0.424119, Arrays.sort: 0.712698

Elementos: 8.000.000, secuencial: 1.279374, roscada: 0.859363, Arrays.sort: 1.487671

Cada número anterior es el tiempo medio de 100 pruebas, tirar los 3 casos más altos más bajos y 3. Solía ??Random.nextInt (Integer.MAX_VALUE) para generar una matriz para cada prueba, que se inicializa una vez cada 10 pruebas con la misma semilla. Cada prueba consistió en medir el tiempo del algoritmo dado con System.nanoTime. Doblé con seis cifras decimales después de promediar. Y, obviamente, lo hice comprobación para ver si cada clase trabajó .

Como se puede ver, no se trata de un aumento del 32% en la velocidad entre el secuencial y casos roscados en cada serie de pruebas. Lo que le pedía anteriormente, no debería espero más que eso?

Answer 1

Hacer numThreads estática puede causar problemas, es muy probable que usted va a terminar con más de max_threads ejecutan en algún momento.

Probablemente la razón por la que no obtiene una completa doble en el rendimiento es que su ordenación rápida no puede ser completamente paralelizado. Tenga en cuenta que la primera llamada a quicksort hará un pase a través de toda la matriz en el hilo inicial antes de que comience a funcionar realmente en paralelo. También hay una sobrecarga en parallelising un algoritmo en forma de transiciones de conmutación contexto y modo cuando la agricultura fuera a hilos separados.

Para consultar el Tenedor / Join marco, este problema sería probablemente encajar muy claramente allí.

Un par de puntos sobre la aplicación. Implementar Ejecutable en lugar de extender Thread. Extensión de un hilo debe ser utilizado sólo cuando se crea una nueva versión de clase Thread. Cuando lo que desea es hacer un poco de trabajo a ser ejecutado en paralelo que está mejor con Ejecutable. Mientras iplementing un Ejecutable también se puede extender aún otra clase que le da más flexibilidad en el diseño orientado a objetos. Utilice un grupo de subprocesos que se limita al número de hilos que tiene disponible en el sistema. Tampoco utilice numThreads para tomar la decisión de si a la mesa de un nuevo hilo o no. Se puede calcular esto desde el principio. Utilice un tamaño mínimo de la partición que es el tamaño de la matriz total dividido por el número de procesadores disponibles. Algo así como:

public class ThreadedQuick implements Runnable {

    public static final int MAX_THREADS = Runtime.getRuntime().availableProcessors();
    static final ExecutorService executor = Executors.newFixedThreadPool(MAX_THREADS);

    final int[] my_array;
    final int start, end;

    private final int minParitionSize;

    public ThreadedQuick(int minParitionSize, int[] array, int start, int end) {
        this.minParitionSize = minParitionSize;
        this.my_array = array;
        this.start = start;
        this.end = end;
    }

    public void run() {
        quicksort(my_array, start, end);
    }

    public void quicksort(int[] array, int start, int end) {
        int len = end - start + 1;

        if (len <= 1)
            return;

        int pivot_index = medianOfThree(array, start, end);
        int pivotValue = array[pivot_index];

        swap(array, pivot_index, end);

        int storeIndex = start;
        for (int i = start; i < end; i++) {
            if (array[i] <= pivotValue) {
                swap(array, i, storeIndex);
                storeIndex++;
            }
        }

        swap(array, storeIndex, end);

        if (len > minParitionSize) {

            ThreadedQuick quick = new ThreadedQuick(minParitionSize, array, start, storeIndex - 1);
            Future<?> future = executor.submit(quick);
            quicksort(array, storeIndex + 1, end);

            try {
                future.get(1000, TimeUnit.SECONDS);
            } catch (Exception ex) {
                ex.printStackTrace();
            }
        } else {
            quicksort(array, start, storeIndex - 1);
            quicksort(array, storeIndex + 1, end);
        }
    }    
}

Se puede poner a raya haciendo:

ThreadedQuick quick = new ThreadedQuick(array / ThreadedQuick.MAX_THREADS, array, 0, array.length - 1);
quick.run();

Esto iniciará el tipo en el mismo hilo, lo que evita un salto de hilo innecesarios en el arranque.

Advertencia:. No estoy seguro de la implementación anterior en realidad será más rápido ya que no he referenciado que

Answer 2

Este utiliza una combinación de tipo de combinación y rápida tipo.

import java.util.Arrays;
import java.util.Random;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ParallelSortMain {
    public static void main(String... args) throws InterruptedException {
        Random rand = new Random();
        final int[] values = new int[100*1024*1024];
        for (int i = 0; i < values.length; i++)
            values[i] = rand.nextInt();

        int threads = Runtime.getRuntime().availableProcessors();
        ExecutorService es = Executors.newFixedThreadPool(threads);
        int blockSize = (values.length + threads - 1) / threads;
        for (int i = 0; i < values.length; i += blockSize) {
            final int min = i;
            final int max = Math.min(min + blockSize, values.length);
            es.submit(new Runnable() {
                @Override
                public void run() {
                    Arrays.sort(values, min, max);
                }
            });
        }
        es.shutdown();
        es.awaitTermination(10, TimeUnit.MINUTES);
        for (int blockSize2 = blockSize; blockSize2 < values.length / 2; blockSize2 *= 2) {
            for (int i = 0; i < values.length; i += blockSize2) {
                final int min = i;
                final int mid = Math.min(min + blockSize2, values.length);
                final int max = Math.min(min + blockSize2 * 2, values.length);
                mergeSort(values, min, mid, max);
            }
        }
    }

    private static boolean mergeSort(int[] values, int left, int mid, int end) {
        int[] results = new int[end - left];
        int l = left, r = mid, m = 0;
        for (; l < left && r < mid; m++) {
            int lv = values[l];
            int rv = values[r];
            if (lv < rv) {
                results[m] = lv;
                l++;
            } else {
                results[m] = rv;
                r++;
            }
        }
        while (l < mid)
            results[m++] = values[l++];
        while (r < end)
            results[m++] = values[r++];
        System.arraycopy(results, 0, values, left, results.length);
        return false;
    }
}

Answer 3

Un par de comentarios si entiendo su código correcto:

1. No veo un bloqueo alrededor de las numthreads objeto a pesar de que se podía acceder a través de varios subprocesos. Tal vez usted debe hacer una AtomicInteger.

2. Utilice un grupo de subprocesos y organizar las tareas, es decir, una sola llamada a la clasificación rápida, para tomar advantange de un grupo de subprocesos. Utilizar futuros.

Su método actual de dividir las cosas de la forma en que está haciendo podría dejar una división más pequeña con un hilo y una división más grande sin un hilo. Es decir, lo hace no prioriza segmentos más grandes con sus propios hilos.