Implementing a Parallel Merge Sort with Phaser

Welcome to Parallel Merge Sort with Phaser

Building on your understanding of concurrency, today we’ll implement a parallel merge sort algorithm using Java's Phaser class. In previous lessons, you explored the nuances of thread-safe data structures like an LRU cache. Now, we’ll take that knowledge further to tackle sorting algorithms in a concurrent environment, optimizing performance on multi-core systems.

What You'll Learn

By the end of this lesson, you will understand: The mechanics of parallel merge sort and how it divides and conquers large datasets. How the Phaser class helps synchronize multiple threads. How to implement an efficient parallel merge sort using thread pools and phased synchronization. This lesson will equip you with the tools to handle large datasets efficiently using parallelism and phased synchronization in Java.

Understanding Parallel Merge Sort with Phaser

Merge sort is a divide-and-conquer algorithm. It breaks down a dataset into smaller subarrays, sorts them, and then merges them back together. The parallel merge sort enhances this process by assigning different subarrays to different threads, allowing multiple parts of the array to be sorted simultaneously. To coordinate these threads, we use Phaser, a synchronization aid that enables threads to wait for each other at specific points, called phases. This is essential in the parallel merge sort, where threads must wait until all chunks are sorted before merging them.

ParallelMergeSort Class Initialization

Here, we define the ParallelMergeSort class. This class manages the overall sorting process, coordinating multiple threads using an ExecutorService for parallel execution and a Phaser for synchronizing the threads. Java public class ParallelMergeSort {private final ExecutorService executor; private final Phaser phaser; private final int numThreads; public ParallelMergeSort(ExecutorService executor, Phaser phaser, int numThreads) {this.executor = executor; this.phaser = phaser; this.numThreads = numThreads;}} In this code, ParallelMergeSort takes three parameters: an ExecutorService, which manages the thread pool, a Phaser for synchronization, and the number of threads. The constructor initializes these fields, preparing the class for sorting. The executor allows tasks to run concurrently across multiple threads, while the phaser ensures that the threads wait for each other at specific points, such as after sorting or merging phases.

Join the 1M+ learners on CodeSignal

Be a part of our community of 1M+ users who develop and demonstrate their skills on CodeSignal

Java

import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Phaser;

public class ParallelMergeSort {

    public void sort(int[] array) throws InterruptedException {
        int chunkSize = (array.length + numThreads - 1) / numThreads;
        phaser.bulkRegister(numThreads); // Register threads with the Phaser

        for (int i = 0; i < numThreads; i++) {
            int left = i * chunkSize;
            int right = Math.min(left + chunkSize - 1, array.length - 1);

            if (left <= right) {
                executor.submit(new SortTask(array, left, right));
            } else {
                phaser.arriveAndDeregister(); // No work to do
            }
        }

        phaser.arriveAndAwaitAdvance(); // Wait for sorting phase to complete
    }
}

import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Phaser;

public class ParallelMergeSort {

    public void sort(int[] array) throws InterruptedException {
        int chunkSize = (array.length + numThreads - 1) / numThreads;
        phaser.bulkRegister(numThreads); // Register threads with the Phaser

        for (int i = 0; i < numThreads; i++) {
            int left = i * chunkSize;
            int right = Math.min(left + chunkSize - 1, array.length - 1);

            if (left <= right) {
                executor.submit(new SortTask(array, left, right));
            } else {
                phaser.arriveAndDeregister(); // No work to do
            }
        }

        phaser.arriveAndAwaitAdvance(); // Wait for sorting phase to complete
    }
}

Java

private class SortTask implements Runnable {
    private final int[] array;
    private final int left;
    private final int right;

    public SortTask(int[] array, int left, int right) {
        this.array = array;
        this.left = left;
        this.right = right;
    }

    @Override
    public void run() {
        Arrays.sort(array, left, right + 1); // Sort the assigned chunk
        phaser.arriveAndDeregister(); // Signal that this task is complete
    }
}

private class SortTask implements Runnable {
    private final int[] array;
    private final int left;
    private final int right;

    public SortTask(int[] array, int left, int right) {
        this.array = array;
        this.left = left;
        this.right = right;
    }

    @Override
    public void run() {
        Arrays.sort(array, left, right + 1); // Sort the assigned chunk
        phaser.arriveAndDeregister(); // Signal that this task is complete
    }
}

Java

public void sort(int[] array) throws InterruptedException {
    // Phase 1: Sort chunks in parallel as shown above...

    int currentChunkSize = chunkSize;
    while (currentChunkSize < array.length) {
        int merges = (array.length + 2 * currentChunkSize - 1) / (2 * currentChunkSize);
        phaser.bulkRegister(merges); // Register merge tasks

        for (int i = 0; i < merges; i++) {
            int left = i * 2 * currentChunkSize;
            int mid = left + currentChunkSize - 1;
            int right = Math.min(left + 2 * currentChunkSize - 1, array.length - 1);

            if (mid < right) {
                executor.submit(new MergeTask(array, left, mid, right));
            } else {
                phaser.arriveAndDeregister(); // No work to do
            }
        }

        phaser.arriveAndAwaitAdvance(); // Wait for merging phase to complete
        currentChunkSize *= 2; // Double chunk size for next phase
    }

    executor.shutdown(); // Shut down the executor service
    System.out.println("Sorted array: " + Arrays.toString(array));
}

public void sort(int[] array) throws InterruptedException {
    // Phase 1: Sort chunks in parallel as shown above...

    int currentChunkSize = chunkSize;
    while (currentChunkSize < array.length) {
        int merges = (array.length + 2 * currentChunkSize - 1) / (2 * currentChunkSize);
        phaser.bulkRegister(merges); // Register merge tasks

        for (int i = 0; i < merges; i++) {
            int left = i * 2 * currentChunkSize;
            int mid = left + currentChunkSize - 1;
            int right = Math.min(left + 2 * currentChunkSize - 1, array.length - 1);

            if (mid < right) {
                executor.submit(new MergeTask(array, left, mid, right));
            } else {
                phaser.arriveAndDeregister(); // No work to do
            }
        }

        phaser.arriveAndAwaitAdvance(); // Wait for merging phase to complete
        currentChunkSize *= 2; // Double chunk size for next phase
    }

    executor.shutdown(); // Shut down the executor service
    System.out.println("Sorted array: " + Arrays.toString(array));
}

Java

private class MergeTask implements Runnable {
    private final int[] array;
    private final int left;
    private final int mid;
    private final int right;

    public MergeTask(int[] array, int left, int mid, int right) {
        this.array = array;
        this.left = left;
        this.mid = mid;
        this.right = right;
    }

    @Override
    public void run() {
        merge(array, left, mid, right);
        phaser.arriveAndDeregister(); // Signal merge completion
    }
}

private class MergeTask implements Runnable {
    private final int[] array;
    private final int left;
    private final int mid;
    private final int right;

    public MergeTask(int[] array, int left, int mid, int right) {
        this.array = array;
        this.left = left;
        this.mid = mid;
        this.right = right;
    }

    @Override
    public void run() {
        merge(array, left, mid, right);
        phaser.arriveAndDeregister(); // Signal merge completion
    }
}

Java

private void merge(int[] array, int left, int mid, int right) {
    int[] leftArr = Arrays.copyOfRange(array, left, mid + 1);
    int[] rightArr = Arrays.copyOfRange(array, mid + 1, right + 1);

    int i = 0, j = 0, k = left;

    while (i < leftArr.length && j < rightArr.length) {
        array[k++] = (leftArr[i] <= rightArr[j]) ? leftArr[i++] : rightArr[j++];
    }
    while (i < leftArr.length) {
        array[k++] = leftArr[i++];
    }
    while (j < rightArr.length) {
        array[k++] = rightArr[j++];
    }
}

private void merge(int[] array, int left, int mid, int right) {
    int[] leftArr = Arrays.copyOfRange(array, left, mid + 1);
    int[] rightArr = Arrays.copyOfRange(array, mid + 1, right + 1);

    int i = 0, j = 0, k = left;

    while (i < leftArr.length && j < rightArr.length) {
        array[k++] = (leftArr[i] <= rightArr[j]) ? leftArr[i++] : rightArr[j++];
    }
    while (i < leftArr.length) {
        array[k++] = leftArr[i++];
    }
    while (j < rightArr.length) {
        array[k++] = rightArr[j++];
    }
}

Java

import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Phaser;

public class Main {

    private static final int ARRAY_SIZE = 16;
    private static final int NUM_THREADS = 4;

    public static void main(String[] args) throws InterruptedException {
        new Main().startParallelMergeSort();
    }

    public void startParallelMergeSort() throws InterruptedException {
        int[] array = new int[ARRAY_SIZE];
        fillRandom(array);
        System.out.println("Unsorted array: " + Arrays

.toString(array));

        ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS);
        Phaser phaser = new Phaser(1); // Main thread is a registered party

        ParallelMergeSort sorter = new ParallelMergeSort(executor, phaser, NUM_THREADS);
        sorter.sort(array);
    }

    private void fillRandom(int[] array) {
        for (int i = 0; i < array.length; i++) {
            array[i] = (int) (Math.random() * 100);
        }
    }
}

import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Phaser;

public class Main {

    private static final int ARRAY_SIZE = 16;
    private static final int NUM_THREADS = 4;

    public static void main(String[] args) throws InterruptedException {
        new Main().startParallelMergeSort();
    }

    public void startParallelMergeSort() throws InterruptedException {
        int[] array = new int[ARRAY_SIZE];
        fillRandom(array);
        System.out.println("Unsorted array: " + Arrays

.toString(array));

        ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS);
        Phaser phaser = new Phaser(1); // Main thread is a registered party

        ParallelMergeSort sorter = new ParallelMergeSort(executor, phaser, NUM_THREADS);
        sorter.sort(array);
    }

    private void fillRandom(int[] array) {
        for (int i = 0; i < array.length; i++) {
            array[i] = (int) (Math.random() * 100);
        }
    }
}

Java

import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Phaser;

public class ParallelMergeSort {

    public void sort(int[] array) throws InterruptedException {
        int chunkSize = (array.length + numThreads - 1) / numThreads;
        phaser.bulkRegister(numThreads); // Register threads with the Phaser

        for (int i = 0; i < numThreads; i++) {
            int left = i * chunkSize;
            int right = Math.min(left + chunkSize - 1, array.length - 1);

            if (left <= right) {
                executor.submit(new SortTask(array, left, right));
            } else {
                phaser.arriveAndDeregister(); // No work to do
            }
        }

        phaser.arriveAndAwaitAdvance(); // Wait for sorting phase to complete
    }
}

Java

private class SortTask implements Runnable {
    private final int[] array;
    private final int left;
    private final int right;

    public SortTask(int[] array, int left, int right) {
        this.array = array;
        this.left = left;
        this.right = right;
    }

    @Override
    public void run() {
        Arrays.sort(array, left, right + 1); // Sort the assigned chunk
        phaser.arriveAndDeregister(); // Signal that this task is complete
    }
}

Java

public void sort(int[] array) throws InterruptedException {
    // Phase 1: Sort chunks in parallel as shown above...

    int currentChunkSize = chunkSize;
    while (currentChunkSize < array.length) {
        int merges = (array.length + 2 * currentChunkSize - 1) / (2 * currentChunkSize);
        phaser.bulkRegister(merges); // Register merge tasks

        for (int i = 0; i < merges; i++) {
            int left = i * 2 * currentChunkSize;
            int mid = left + currentChunkSize - 1;
            int right = Math.min(left + 2 * currentChunkSize - 1, array.length - 1);

            if (mid < right) {
                executor.submit(new MergeTask(array, left, mid, right));
            } else {
                phaser.arriveAndDeregister(); // No work to do
            }
        }

        phaser.arriveAndAwaitAdvance(); // Wait for merging phase to complete
        currentChunkSize *= 2; // Double chunk size for next phase
    }

    executor.shutdown(); // Shut down the executor service
    System.out.println("Sorted array: " + Arrays.toString(array));
}

Java

private class MergeTask implements Runnable {
    private final int[] array;
    private final int left;
    private final int mid;
    private final int right;

    public MergeTask(int[] array, int left, int mid, int right) {
        this.array = array;
        this.left = left;
        this.mid = mid;
        this.right = right;
    }

    @Override
    public void run() {
        merge(array, left, mid, right);
        phaser.arriveAndDeregister(); // Signal merge completion
    }
}

Java

private void merge(int[] array, int left, int mid, int right) {
    int[] leftArr = Arrays.copyOfRange(array, left, mid + 1);
    int[] rightArr = Arrays.copyOfRange(array, mid + 1, right + 1);

    int i = 0, j = 0, k = left;

    while (i < leftArr.length && j < rightArr.length) {
        array[k++] = (leftArr[i] <= rightArr[j]) ? leftArr[i++] : rightArr[j++];
    }
    while (i < leftArr.length) {
        array[k++] = leftArr[i++];
    }
    while (j < rightArr.length) {
        array[k++] = rightArr[j++];
    }
}

Java

import java.util.Arrays;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Phaser;

public class Main {

    private static final int ARRAY_SIZE = 16;
    private static final int NUM_THREADS = 4;

    public static void main(String[] args) throws InterruptedException {
        new Main().startParallelMergeSort();
    }

    public void startParallelMergeSort() throws InterruptedException {
        int[] array = new int[ARRAY_SIZE];
        fillRandom(array);
        System.out.println("Unsorted array: " + Arrays

.toString(array));

        ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS);
        Phaser phaser = new Phaser(1); // Main thread is a registered party

        ParallelMergeSort sorter = new ParallelMergeSort(executor, phaser, NUM_THREADS);
        sorter.sort(array);
    }

    private void fillRandom(int[] array) {
        for (int i = 0; i < array.length; i++) {
            array[i] = (int) (Math.random() * 100);
        }
    }
}

Welcome to Parallel Merge Sort with Phaser

What You'll Learn

Understanding Parallel Merge Sort with Phaser

ParallelMergeSort Class Initialization

Join the 1M+ learners on CodeSignal

Sorting Chunks in Parallel

SortTask Class

Merging Sorted Chunks

MergeTask Class

Merge Method

Running the Program

Why Parallel Merge Sort with Phaser Matters