Java: parallelizzare quick sort tramite multi-threading

Question

Sto sperimentando con parallelizzazione algoritmi in Java. Ho iniziato con merge sort, e inviato il mio tentativo in questa domanda . Il mio tentativo è rivisto nel seguente codice, dove ora cerco di parallelizzare quick sort.

Ci sono errori rookie nella mia implementazione multi-threaded o approccio a questo problema? In caso contrario, non dovrei aspettare più di un aumento di velocità del 32% tra un sequenziale e un algoritmo parallelizzato su un duello-core (vedi temporizzazioni in basso)?

Ecco l'algoritmo di multithreading:

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

Ecco come mi metto fuori:

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

Ho provato questo contro Arrays.sort e un algoritmo di ordinamento rapido sequenziale simile. Ecco i risultati di temporizzazione su un dell laptop Intel duello-core, in secondi:

Elementi: 500.000, sequenziale: 0.068592, filettato: 0.046871, Arrays.sort: 0.079677

Elementi: 1.000.000, sequenziale: 0,14,416 mila, filettato: 0.095492, Arrays.sort: 0.167155

Elementi: 2.000.000, sequenziale: 0.301666, filettato: 0.205719, Arrays.sort: 0.350982

Elementi: 4.000.000, sequenziale: 0.623291, filettato: 0.424119, Arrays.sort: 0.712698

Elementi: 8.000.000, sequenziale: 1.279374, filettato: 0.859363, Arrays.sort: 1.487671

Ogni numero di cui sopra è il tempo medio di 100 test, buttare fuori i 3 più bassi e più alti 3 casi. Ho usato Random.nextInt (Integer.MAX_VALUE) per generare una matrice per ogni prova, che è stata inizializzata una volta ogni 10 prove con lo stesso seme. Ogni prova consisteva temporizzazione dell'algoritmo proposta con System.nanoTime. Ho arrotondato a sei cifre decimali, dopo una media. E ovviamente, ho controllato per vedere se ogni sorta ha lavorato .

Come si può vedere, non v'è circa un aumento del 32% in termini di velocità tra il sequenziale e casi filettati in ogni serie di test. Come ho chiesto in precedenza, non dovrebbe mi aspetto di più?

Answer 1

Fare numThreads statica può causare problemi, è molto probabile che si finirà con più di MAX_THREADS in esecuzione ad un certo punto.

Probabilmente la ragione per la quale non si ottiene un doppio up completo in termini di prestazioni è che il vostro quick sort non può essere completamente parallelised. Si noti che la prima chiamata a quicksort farà un passaggio attraverso l'intera matrice del filo iniziale prima che decorre realmente in parallelo. C'è anche un overhead in Parallelizzazione un algoritmo in forma di commutazione contesto e modalità transizioni quando l'agricoltura off per thread separati.

Date un'occhiata al bivio / Join quadro, questo problema sarebbe probabilmente in forma abbastanza ordinatamente lì.

Un paio di punti sulla realizzazione. Implementare Runnable piuttosto che si estende Thread. Estensione di una discussione deve essere utilizzato solo quando si crea una nuova versione della classe Thread. Quando si desidera solo per fare un po 'di lavoro da eseguire in parallelo si sta meglio con Runnable. Mentre iplementing un Runnable è anche possibile estendere ancora un'altra classe, che offre una maggiore flessibilità nella progettazione OO. Utilizzare un pool di thread che è limitato al numero di thread che avete a disposizione nel sistema. Inoltre, non utilizzare numThreads prendere la decisione sull'opportunità di un fork di un nuovo thread o meno. È possibile calcolare questo fronte. Utilizzare una dimensione minima della partizione che è la dimensione della matrice totale diviso per il numero di processori disponibili. Qualcosa di simile:

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

Si può cacciare fuori facendo:

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

Questo farà partire il genere nello stesso thread, che evita un inutile hop filo all'avvio.

Avvertimento:. Non sono sicuro quanto sopra implementazione sarà effettivamente più veloce in quanto non ho benchmark che

Answer 2

Questo utilizza una combinazione di rapido ordinamento e merge sort.

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 paio di commenti, se capisco il vostro codice a destra:

1. Non vedo un blocco intorno alle numthreads oggetto anche se potrebbe essere accessibile tramite più thread. Forse si dovrebbe fare un AtomicInteger.

2. Usa un pool di thread e di organizzare i compiti, vale a dire una singola chiamata a quicksort, a prendere advantange di un pool di thread. Utilizzare i futures.

Il tuo attuale metodo di dividere le cose nel modo che stai facendo potrebbe lasciare una divisione più piccola con un filo e una divisione più grande senza un filo. Vale a dire, lo fa segmenti non Dare priorità più grandi con le loro discussioni.