Scale Up Your Code With Java Concurrency

As you've seen, breaking your workload into smaller parts can make it faster. You've also seen when using locks and atomic types, that it's easy to lose the benefits of concurrency by forcing sequential execution of critical sections. So far, we've played it safe by using two threads at a time. Could we break the work down further? Perhaps by splitting the input into lots of smaller files that are all processed concurrently, before averaging a final result?

In this chapter, you're going to learn to create the threads you need while avoiding an explosion of too many threads running concurrently. This saves you from using up all your memory or getting stuck in a state where the scheduler has too many threads to run. We call this situation starvation, as threads might starve from lack of work. 🍽

Setting Thread Limits Starting at One: Semaphores

Have you ever been stopped by a traffic light at a pedestrian crossing, along with a bunch of other people?

OK, but why did it stop you?

Exactly! In Java concurrency code, a semaphore serves as a kind of traffic light. It allows you to throw lots of threads at a problem while restricting the number working on your critical sections at any given time.

A semaphore, just like a lock, is designed to allow threads to coordinate when your design requires restrictions on the amount of concurrent work done on critical sections of code.

It can be used to limit the total number of workers (threads) solving a problem at the same time, ensuring that you manage shared resources fairly.

For instance, if you have workers querying a web-service, you might want to limit the number of threads that can interact with it at any time. If you have too many threads querying it, you might add more load onto the service than was agreed - and cause it to crash. So you'd only grant a limited number of "access permits." Threads with a permit would begin work, and as each of them terminates, they pass on their permit to new threads.

When you create a semaphore, you do so with a limit. If any more workers (threads) want to help out, they'll need to wait for the number of active workers to fall below the limit again. The three important steps involved in using a semaphore conveniently spell out the word CAR:

1. Create a semaphore and state how many permits it should have.

2. Acquire a permit from your semaphore when you want a thread to proceed. The thread might have to wait until a permit is available.

3. Release a permit when your thread has completed the work you wanted to restrict.

The most basic semaphore you can create has a limit of 1, meaning that only 1 thread at a time may acquire a semaphore and proceed to run the code it guards. It's essentially a lock. You call this a binary semaphore, as it always has 1 or 0 permits available. That means that there can only be one! 🗡

Java's concurrency framework provides you with the with java.util.concurrent.Semaphore class. You can use this to create a binary semaphore. Let's do this to ensure that only one person can sing at a time. This is useful in avoiding harmony-related headaches. 🤕

import java.util.concurrent.Semaphore;

public class Singer extends Thread {
    // Anly allow one person to sing at a time
    static Semaphore semaphore = new Semaphore(1);
    

    public void run() {
        try {
            // Aquire a permit to proceed
            semaphore.acquire();
            System.out.println("DoReMiFaSoLaTiDo!");
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // Always release our permit
            semaphore.release();
        }
    }
}

• Line 5: Create a shared instance of a semaphore with a limit of 1.

• Line 11: Acquire a semaphore, the same way we called lock() previously. When a permit is available, this allows the thread to run its critical section at Line 12. OK, it just prints a line, but you'd find it critical if you were performing in the Sound of Music!

• Line 18: Release the semaphore in a finally block. As this is guaranteed to run after try and catch blocks, it ensures that you always release the semaphore. You'll note that we used this same pattern for ReenterantLock::unlock.

Try It Out for Yourself!

Use the code to start a band where only one person can sing at a time.

Step 1: Start up JShell:

jshell

Step 2:  Paste the code from above for the singer class into JShell.

Step 3: Create four singer threads named john, paul, ringo, and george in JShell.

jshell> Singer john = new Singer();
john ==> Thread[Thread-4,5,main]

jshell> Singer paul = new Singer();
paul ==> Thread[Thread-5,5,main]

jshell> Singer ringo = new Singer();
ringo ==> Thread[Thread-6,5,main]

jshell> Singer george = new Singer();
george ==> Thread[Thread-7,5,main]

Step 4:  Start each thread.

jshell> john.start(); paul.start(); ringo.start(); george.start()
DoReMiFaSoLaTiDo!
DoReMiFaSoLaTiDo!
DoReMiFaSoLaTiDo!
DoReMiFaSoLaTiDo!

Call start, and each, in turn, sings its line. Now that they are singing individually, it's not hard to get them to take turns.

One of the hardest things in modern concurrent programming is deciding on the correct concurrency APIs to use. An excellent way to understand when to use any API is to understand the situations it is designed to address. Semaphores are designed to make sure that you can fairly share resources, such as the previous example relating to a web server. Here are a few rules of thumb:

• Do you need to be able to easily adjust the number of threads that do a particular job at the same time?

• Will having too many concurrent threads result in overworking a shared resource, such as a mail server or weather sensor?

• Will you have lots of short-lived threads that start and stop? Semaphores allow you to have lots of threads created, but make sure that only a fixed number are ever active at the same time.

Now that you've got the basics down, you're ready to add some room in here for a few more semaphores! 😉

Increasing Your Thread Limit: n-Semaphores

Semaphores work best when you can use them to help get the benefits of multiple threads working together, but easily restrict how many threads can have permits to proceed.

It's a little like a parking lot, where you have a fixed number of parking permits available. A parking lot with 20 parking spots is never able to fit 21 cars. That 21st car has to wait for another car to leave and surrender its permit. Similarly, individual threads may start and stop, but the total number never exceeds the limit you set. Additional threads go into  WAIT state until they can acquire a permit.

To turn a binary semaphore into a semaphore with more permits, you pass the constructor some larger number. For example, the following creates a semaphore that only allows two permits at any time.

Semaphore semaphore = new Semaphore(2);

Making this change to the singer class from before results in all four singers singing together.

Creating a Semaphore That Allows Two Singers to Harmonize

Step 1: Startup JShell. Recreate your singer class with some exceptions, which help you see what it's doing.

Step 2: Change the number of semaphore permits to 2. 

...
static Semaphore semaphore = new Semaphore(2);
...

Step 3: Add a custom constructor to the class, which takes in the name of the singer. This helps you report on what your thread is doing.

public class Singer extends Thread {
...
    private String singer;
    
    public Singer(String singer) {
        super();
        this.singer = singer;
    }
    
...
}

Step 4: Add a method to report on the number of available permits:

...
    private static void reportPermits() {
        System.out.println(
            "There are " + semaphore.availablePermits() 
            + " permits available"); 
    }
...

This uses the availablePermits() method.

Step 5:  Add some reporting, so the whole class now looks like:

import java.util.concurrent.Semaphore;

public class Singer extends Thread {
    // Anly allow two people to sing at a time
    static Semaphore semaphore = new Semaphore(2);
    private String singer;
    
    public Singer(String singer) {
        super();
        this.singer = singer;
    }
    
    private static void reportPermits() {
        System.out.println(
            "There are " + semaphore.availablePermits() 
            + " permits available"); 
    }

    public void run() {
        try {
            // number of permits before acquire
            reportPermits();
            
            // Aquire a permit to proceed
            semaphore.acquire();
            
            // Take a deep breadth
            Thread.sleep(500);
            
            // Sing
            System.out.println(singer + " : DoReMiFaSoLaTiDo!");
            
            // Number of permits during singing
            reportPermits();
            
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            // Always release our permit
            semaphore.release();
        }
        
        // Number of permits after release()
        reportPermits();
    }
}

The main differences are:

• Lines 3 to 8: The constructor now receives the name of a singer.

• Lines 22, 44, and 34: Report the number of available semaphore permits before, after, and during singing.

• Line 28: A Thread.sleep(500) makes each singer take a breath for half a second. It makes things a little more unpredictable, and it gives you some variety in your output.

Step 6: Create four singer threads named john, paul, ringo, and george in JShell.

jshell> Singer lead = new Singer("John");
lead ==> Thread[Thread-25,5,main]

jshell> Singer bass = new Singer("Paul");
bass ==> Thread[Thread-26,5,main]

jshell> Singer drums = new Singer("Ringo");
drums ==> Thread[Thread-27,5,main]

jshell> Singer rhythm = new Singer("George");
rhythm ==> Thread[Thread-28,5,main]

 Step 7: Start the threads!

jshell> lead.start(); bass.start(); drums.start(); rhythm.start();

Here's what I get when I let them sing in pairs. 🎵

There are 2 permits available
There are 1 permits available
There are 0 permits available
John : DoReMiFaSoLaTiDo!
There are 0 permits available
There are 1 permits available
There are 0 permits available

jshell> Paul : DoReMiFaSoLaTiDo!
There are 0 permits available
There are 1 permits available
Ringo : DoReMiFaSoLaTiDo!
There are 0 permits available
There are 1 permits available
George : DoReMiFaSoLaTiDo!
There are 1 permits available
There are 2 permits available

As you can see, the order of available permits changes as each thread starts and stops singing. At the very beginning, however, there are two permits, and that is what you end with.

That is, the semaphore makes sure that you always have up to two permits available.

Updating Our Planet Analyzer to Use Semaphores: Practice!

I've modified the planet analyzer to use an n-semaphore. The 100, 000+ planet records previously used have now been split across 23 files of about 5000 each. Using an n-semaphore of 8 allows individual threads to process each of these files concurrently. The semaphore makes sure that you don't create too many threads reading off the file system at the same time. Remember, semaphores are there to ensure you safely share resources.

There's no right or wrong answer; you'll find this through trial and error. As you'll see, when I walk you through the code, sometimes you have to set a value and see how it impacts the performance of your application. Also, remember that your scheduler keeps switching between threads, while only being able to, at any moment, run as many threads as you have cores.

Now, let's see the code and run the benchmark!

As you saw, this implementation reuses the ThreadBasedPlanetFileAnalyzer service previously created to take a path and return the sum of temperatures and a sample count. It works without using any shared-mutable variables.

The two main pieces of code that differentiated this implementation from others, such as ReentrantLocks were:

• Creating futures to sum and count the sample size for each file.

• Acquiring and releasing up to 8 semaphores to restrict threads from the pool of 20, which concurrently read from the file system.

The method summing the temperatures took the following structure:

// Create Semaphore of Size 8
static Semaphore semaphore = new Semaphore(8);
...
private static Double sumFileWithSemaphores(Path path) throws InterruptedException, IOException {
    
    Double result = null;
    try {
        // ACQUIRE A SEMAPHORE
        semaphore.acquire();
        result = fileAnalyzer.sumFile(path);
    } finally {
        // RELEASE A SEMAPHORE
        semaphore.release();
    }
    return result;
}

• Line 2: Creates a Semaphore with eight permits.

• Line 9: The thread attempts to acquire a permit, either getting one back immediately or having to wait until another is released.

• Line 10: The fileAnalyzer service calculates the result of summing a particular file. There are only eight of these at work at any time, thanks to the semaphore. 

• Line 13: Releases the semaphore, allowing at most, eight threads to proceed past Line 10.

Once you've created all your threads, you wait for them to terminate by calling .get()  on each future, and adding up their results in the parent thread:

// 6 Find the total sum
Double sumOfTemperatures = 0.0;
for (Future<Double> future : futuresListOfFileSums) {
    sumOfTemperatures += future.get();
}

• Line 4: As each result becomes available, add it to the subtotal, to use later in calculating an average.

Running the Benchmark!

See how the benchmark performs on your machine:

git checkout p2-c2-semaphores

And then run the benchmarks with the runBenchmarks gradle task:

./gradlew runBenchmarks

How did it compare? Did you try to modify the number of semaphores and the size of the thread pool and rerun the benchmarks? Also, try setting it to two semaphores and reducing the thread pool size to four. How does that perform? 😀

Let's Recap!

• Use semaphores to set a limit on the number of threads performing the same task at one time. Use semaphores to help share resources fairly, which would otherwise affect your application's performance or its obligations when using external resources, like another microservice.

• Semaphores have a fixed number of permits that a thread can request and acquire. It releases it when it's done with your contentious resource.

  • java.util.concurrent.Semaphore provides a constructor that takes a value specifying the number of permits to associate with that semaphore instance.

  • A thread calling a particular semaphore's acquire()  method attempts to obtain a permit. It proceeds if one is available; otherwise, it blocks and waits until one is released.

• Any thread acquiring a semaphore is responsible for releasing it - ideally in a finally block to ensure that an exception does not prevent threads from releasing their permit.

• A binary semaphore behaves like a lock, as it has a limit of 1. This provides exclusive access to code guarded by a semaphore. It may be preferable to a lock if you expect to increase that limit one day.

 In the next chapter, we'll look at how to add a useful feature to our code: countdown latches!