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

Sorting Chunks in Parallel

The first step in parallel merge sort is dividing the array into smaller chunks and sorting each chunk in parallel. This is done using an executor service that distributes tasks among multiple threads. The Phaser class ensures that threads synchronize during different phases, such as sorting and merging.

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

In this part of the code, the array is divided into smaller chunks, and each chunk is assigned to a thread for sorting. The bulkRegister() method is used to register all threads with the Phaser so that they synchronize during each phase of the algorithm. Each chunk is sorted by submitting a SortTask to the ExecutorService, allowing the tasks to run in parallel. The arriveAndAwaitAdvance() method ensures that no thread moves to the next phase until all sorting is completed.

Merging Sorted Chunks

Once all chunks are sorted, the next phase begins: merging the sorted chunks. This merging process occurs in several steps, starting with small chunks and progressively merging larger ones, until the entire array is sorted.

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

In this section, the sorted chunks are merged together in a phased manner. The process starts with the smallest chunks and proceeds to merge larger and larger ones. The Phaser ensures that no thread moves to the next phase until all merges in the current phase are completed. The executor.submit() method runs each merge in parallel, and the process continues until the entire array is sorted.

Running the Program

Here’s how the program is executed from the Main class:

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

In this Main class, we create a random array of integers and print it before and after sorting. The ParallelMergeSort class is instantiated with a fixed thread pool and a Phaser instance. The startParallelMergeSort() method initiates the parallel merge sort process.

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