Parallel Divide-and-Conquer

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83

import java.util.concurrent.*;
import java.util.Arrays;
 
public class ParallelMergeSort extends RecursiveAction {
    private static final int THRESHOLD = 4096;
    
    private final int[] array, buffer;
    private final int lo, hi;
    
    public ParallelMergeSort(int[] array, int[] buffer, int lo, int hi) {
        this.array = array;
        this.buffer = buffer;
        this.lo = lo;
        this.hi = hi;
    }
    
    @Override
    protected void compute() {
        int size = hi - lo;
        
        // Base case: use sequential sort
        if (size <= THRESHOLD) {
            Arrays.sort(array, lo, hi);
            return;
        }
        
        int mid = lo + size / 2;
        
        // DIVIDE & CONQUER: Sort halves in parallel
        ParallelMergeSort left = new ParallelMergeSort(array, buffer, lo, mid);
        ParallelMergeSort right = new ParallelMergeSort(array, buffer, mid, hi);
        
        invokeAll(left, right);  // Fork both, join both
        
        // COMBINE: Merge sorted halves
        merge(lo, mid, hi);
    }
    
    private void merge(int lo, int mid, int hi) {
        // Copy to buffer
        System.arraycopy(array, lo, buffer, lo, hi - lo);
        
        int i = lo, j = mid, k = lo;
        while (i < mid && j < hi) {
            if (buffer[i] <= buffer[j]) {
                array[k++] = buffer[i++];
            } else {
                array[k++] = buffer[j++];
            }
        }
        while (i < mid) array[k++] = buffer[i++];
        while (j < hi) array[k++] = buffer[j++];
    }
    
    public static void sort(int[] array) {
        int[] buffer = new int[array.length];
        ForkJoinPool.commonPool().invoke(
            new ParallelMergeSort(array, buffer, 0, array.length));
    }
    
    public static void main(String[] args) {
        int[] data = new int[10_000_000];
        for (int i = 0; i < data.length; i++) {
            data[i] = (int)(Math.random() * 1_000_000);
        }
        
        long start = System.nanoTime();
        sort(data);
        long elapsed = System.nanoTime() - start;
        
        System.out.printf("Sorted %d elements in %.2f ms%n",
                          data.length, elapsed / 1e6);
        
        // Verify
        for (int i = 1; i < data.length; i++) {
            if (data[i] < data[i-1]) {
                System.out.println("ERROR: Not sorted!");
                return;
            }
        }
        System.out.println("Verified: Array is sorted");
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

import java.util.concurrent.*;
import java.util.Arrays;
 
public class ParallelQuicksort extends RecursiveAction {
    private static final int THRESHOLD = 4096;
    
    private final int[] array;
    private final int lo, hi;
    
    public ParallelQuicksort(int[] array, int lo, int hi) {
        this.array = array;
        this.lo = lo;
        this.hi = hi;
    }
    
    @Override
    protected void compute() {
        if (hi - lo <= THRESHOLD) {
            Arrays.sort(array, lo, hi);
            return;
        }
        
        // DIVIDE: Partition around pivot
        int pivotIndex = partition(lo, hi);
        
        // CONQUER: Sort partitions in parallel
        ParallelQuicksort left = new ParallelQuicksort(array, lo, pivotIndex);
        ParallelQuicksort right = new ParallelQuicksort(array, pivotIndex + 1, hi);
        
        // Both partitions are independent - parallelize
        invokeAll(left, right);
        
        // COMBINE: Nothing to do - array is sorted in place
    }
    
    private int partition(int lo, int hi) {
        // Median-of-three pivot selection for better balance
        int mid = lo + (hi - lo) / 2;
        if (array[mid] < array[lo]) swap(lo, mid);
        if (array[hi-1] < array[lo]) swap(lo, hi-1);
        if (array[hi-1] < array[mid]) swap(mid, hi-1);
        
        int pivot = array[mid];
        swap(mid, hi - 1);  // Move pivot to end
        
        int i = lo, j = hi - 2;
        while (true) {
            while (array[i] < pivot) i++;
            while (j > lo && array[j] > pivot) j--;
            if (i >= j) break;
            swap(i++, j--);
        }
        swap(i, hi - 1);  // Restore pivot
        return i;
    }
    
    private void swap(int i, int j) {
        int temp = array[i];
        array[i] = array[j];
        array[j] = temp;
    }
    
    public static void sort(int[] array) {
        ForkJoinPool.commonPool().invoke(
            new ParallelQuicksort(array, 0, array.length));
    }
}