Concurrent Image Processing Pipeline

Introduction to Concurrent Image Processing Pipeline

Welcome back! In the previous lesson, we explored how to design a concurrent garbage collector simulation, focusing on memory management and thread coordination. Today, we’ll pivot toward another powerful application of concurrency — Concurrent Image Processing Pipelines . This concept builds on your foundational understanding of concurrency and multithreading to process images more efficiently.

What You'll Learn

In this lesson, you will: Improve your understanding of parallel processing and task coordination using Java . Learn to use the Phaser class for synchronizing tasks in multiple phases. Understand how concurrent processing can optimize performance on multi-core systems. By the end of this lesson, you'll be able to design a multi-phase image processing pipeline that applies several image filters concurrently, optimizing execution on multiple threads.

Building a Concurrent Image Processing Pipeline

The core idea of a concurrent image processing pipeline is to divide an image into equal parts and process these parts simultaneously across multiple threads. This can significantly speed up processing tasks, especially for large images. We use a Phaser to synchronize these threads, ensuring each phase is complete before moving on to the next. Let’s quickly recall the Phaser class in Java, which is ideal for managing tasks that need to be synchronized across multiple phases. Unlike other synchronization mechanisms like CountDownLatch or CyclicBarrier, the Phaser is more flexible, allowing a variable number of threads to join or leave during execution. This makes it well-suited for dynamic environments where tasks might need to adapt or scale.

Setting Up the Image Processor

Our solution involves an ImageProcessor class responsible for applying filters to sections of an image. Each thread handles a specific segment, applying a series of transformations (filters) in a coordinated manner. Java import java.util.concurrent.Phaser; public class ImageProcessor implements Runnable {private final int[] image; private final int start; private final int end; private final Phaser phaser; public ImageProcessor(int[] image, int start, int end, Phaser phaser) {this.image = image; this.start = start; this.end = end; this.phaser = phaser;}} import java.util.concurrent.Phaser; public class ImageProcessor implements Runnable {private final int[] image; private final int start; private final int end; private final Phaser phaser; public ImageProcessor(int[] image, int start, int end, Phaser phaser) {this.image = image; this.start = start; this.end = end; this.phaser = phaser;}} In the above snippet, the ImageProcessor takes in an image array and processes a segment of the image specified by the start and end indices. Each thread operates on its section of the image, coordinating with others using the Phaser . This design allows us to divide the workload among multiple threads, with each responsible for a distinct part of the image.

Join the 1M+ learners on CodeSignal

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

Processing Image Phases

Now, let’s implement the logic that applies multiple filters in phases, ensuring synchronization across threads using the Phaser.

Java

@Override
public void run() {
    try {
        // Phase 1: Apply filter 1
        for (int i = start; i < end; i++) {
            image[i] += 1; // Simulate filter
        }
        System.out.println("Thread " + Thread.currentThread().getName() + " completed Phase 1");
    } finally {
        phaser.arriveAndAwaitAdvance();
    }

    try {
        // Phase 2: Apply filter 2
        for (int i = start; i < end; i++) {
            image[i] *= 2; // Simulate filter
        }
        System.out.println("Thread " + Thread.currentThread().getName() + " completed Phase 2");
    } finally {
        phaser.arriveAndAwaitAdvance();
    }

    try {
        // Phase 3: Apply filter 3
        for (int i = start; i < end; i++) {
            image[i] -= 3; // Simulate filter
        }
        System.out.println("Thread " + Thread.currentThread().getName() + " completed Phase 3");
    } finally {
        phaser.arriveAndDeregister();
    }
}

@Override
public void run() {
    try {
        // Phase 1: Apply filter 1
        for (int i = start; i < end; i++) {
            image[i] += 1; // Simulate filter
        }
        System.out.println("Thread " + Thread.currentThread().getName() + " completed Phase 1");
    } finally {
        phaser.arriveAndAwaitAdvance();
    }

    try {
        // Phase 2: Apply filter 2
        for (int i = start; i < end; i++) {
            image[i] *= 2; // Simulate filter
        }
        System.out.println("Thread " + Thread.currentThread().getName() + " completed Phase 2");
    } finally {
        phaser.arriveAndAwaitAdvance();
    }

    try {
        // Phase 3: Apply filter 3
        for (int i = start; i < end; i++) {
            image[i] -= 3; // Simulate filter
        }
        System.out.println("Thread " + Thread.currentThread().getName() + " completed Phase 3");
    } finally {
        phaser.arriveAndDeregister();
    }
}

Here, the image processing occurs in three phases, simulating the application of three filters. In Phase 1, each pixel in the image segment is incremented by 1. In Phase 2, the pixel values are doubled, and in Phase 3, 3 is subtracted from each value. Each phase ensures that all threads complete their tasks before moving on to the next by calling phaser.arriveAndAwaitAdvance(). After the final phase, the thread deregisters from the Phaser using arriveAndDeregister() to indicate it has finished.

Coordinating Threads Using the Main Class

The Main class coordinates thread execution, dividing the image array among a fixed number of threads and initiating the processing.

Java

import java.util.Arrays;
import java.util.concurrent.*;

public class Main {

    private static final int IMAGE_SIZE = 1000;
    private static final int NUM_THREADS = 4;

    public static void main(String[] args) throws InterruptedException {
        int[] image = new int[IMAGE_SIZE];
        Arrays.fill(image, 1); // Simulate an image with pixel value 1

        ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS);
        Phaser phaser = new Phaser(NUM_THREADS);

        int chunkSize = IMAGE_SIZE / NUM_THREADS;

import java.util.Arrays;
import java.util.concurrent.*;

public class Main {

    private static final int IMAGE_SIZE = 1000;
    private static final int NUM_THREADS = 4;

    public static void main(String[] args) throws InterruptedException {
        int[] image = new int[IMAGE_SIZE];
        Arrays.fill(image, 1); // Simulate an image with pixel value 1

        ExecutorService executor = Executors.newFixedThreadPool(NUM_THREADS);
        Phaser phaser = new Phaser(NUM_THREADS);

        int chunkSize = IMAGE_SIZE / NUM_THREADS;

In this snippet, the image array is initialized, simulating an image where all pixel values are 1. The ExecutorService manages the thread pool, while the Phaser ensures that the processing happens in phases. The image is divided into chunks, with each thread responsible for processing a specific chunk.

Java

        for (int i = 0; i < NUM_THREADS; i++) {
            int start = i * chunkSize;
            int end = (i == NUM_THREADS - 1) ? IMAGE_SIZE : start + chunkSize;
            executor.submit(new ImageProcessor(image, start, end, phaser));
        }

        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);

        for (int i = 0; i < NUM_THREADS; i++) {
            int start = i * chunkSize;
            int end = (i == NUM_THREADS - 1) ? IMAGE_SIZE : start + chunkSize;
            executor.submit(new ImageProcessor(image, start, end, phaser));
        }

        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);

In this section, the ExecutorService submits tasks to process each chunk of the image concurrently. Each thread processes its assigned chunk in multiple phases, synchronized by the Phaser, ensuring that no phase starts until all threads have completed the previous phase.

Introduction to Concurrent Image Processing Pipeline

What You'll Learn

Building a Concurrent Image Processing Pipeline

Setting Up the Image Processor

Join the 1M+ learners on CodeSignal

Processing Image Phases

Coordinating Threads Using the Main Class

Verifying the Results

Why It Matters