Java programming language

Concurrent programming

Threads and shared data

In this subsection we are going to see which types of objects and operations we can use in order to deal with simple values or collections safely in a multi-threaded application. We will also distinguish between making a collection thread-safe and making the objects stored inside it thread-safe.

1. Accessing single values

Let’s start by learning how to solve some problems associated with single values (primitive and object), how to write (update its value) and read them safely when many threads are accessing at the same time.

1.1. Introducing the problem

Let’s suppose that we have a class to manage a variable (integer) and some methods to modify it and get its value.

public class SimpleInteger 
{
    int num;

    public SimpleInteger(int num) 
    {
        this.num = num;
    }

    public int getNum() 
    {
        return num;
    }

    public void setNum(int num) 
    {
        this.num = num;
    }

    public void increment() 
    {
        num++;
    }
}

Now, let’s create many tasks that modify the value of that number by calling the available increment method. We will manage these tasks with an ExecutorService:

public static SimpleInteger simpleInt = new SimpleInteger(0);

public static void main(String[] args)
{
    ExecutorService executor = Executors.newCachedThreadPool();

    for (int i = 0; i < 10000; i++) 
    {
        executor.execute(() -> simpleInt.increment());
    }

    executor.shutdown();

    try
    {
        if (!executor.awaitTermination(1, TimeUnit.MINUTES))
            executor.shutdownNow();
    }
    catch (InterruptedException e)
    {
        executor.shutdownNow();
        Thread.currentThread().interrupt();
    }

    System.out.println("Expected: 10000, Result: " + simpleInt.getNum());
}

Notice that we submit 10,000 tasks, not necessarily 10,000 threads. The executor decides how many platform threads are actually used to execute those tasks.

What happens when we execute this code? We may have results like these:

Expected: 10000, Result: 9954
Expected: 10000, Result: 9924

The exact result depends on the execution, and in some runs we could even obtain 10,000 by chance. However, the code is not thread-safe.

The problem is that num++ is a read-modify-write operation and is not atomic. Conceptually, it involves these steps:

  1. Read the current value of num.
  2. Calculate the new value.
  3. Write the new value back to num.

These are conceptual steps; the compiler and processor are free to implement the operation differently.

Imagine that the current value is 5. Thread A reads 5 and calculates 6. Before it stores 6, thread B may also read 5 and calculate 6. If both threads finally write 6, one increment has been lost.

This is called a race condition: the final result depends on the relative timing of multiple threads accessing shared mutable data.

1.2. Solving it with synchronization

As we have seen in previous documents, a synchronized block or method uses the monitor associated with an object. Only one thread at a time can execute code protected by the same monitor. A thread that tries to enter while another thread owns that monitor becomes blocked until the monitor is released.

Besides mutual exclusion, entering and leaving synchronized code also provides the corresponding memory-visibility guarantees between threads.

We can protect the increment operation in different ways:

executor.execute(() -> {
    synchronized(simpleInt) 
    {
        simpleInt.increment();
    }
});

public void increment() 
{
    synchronized(this) 
    {
        num++;
    }
}

public synchronized void increment() 
{
    num++;
}

For this example, we only need one of these solutions, as long as every access that must be mutually exclusive uses the same synchronization strategy and monitor.

Now the increment operation is protected, and after every submitted task finishes the result is 10,000.

1.3. Solving it with atomic objects

Instead of explicit synchronization, for many simple operations we can use the classes in the java.util.concurrent.atomic package. These classes provide thread-safe atomic operations on single variables and are commonly used for counters, flags and references.

For instance, instead of doing this:

long num = 10;
num++;

we can use an AtomicLong:

AtomicLong num = new AtomicLong(10);
num.incrementAndGet();

AtomicInteger and AtomicLong provide operations such as incrementAndGet, getAndIncrement, addAndGet, getAndAdd and compareAndSet.

If we apply this to our problem, we can use an AtomicInteger instead of a plain int:

public class SimpleInteger
{
    private final AtomicInteger num;

    public SimpleInteger(int num) 
    {
        this.num = new AtomicInteger(num);
    }

    public int getNum() 
    {
        return num.get();
    }

    public void setNum(int num) 
    {
        this.num.set(num);
    }

    public void increment() 
    {
        num.incrementAndGet();
    }
}

There is also an AtomicReference<T> class that lets us atomically read, replace or compare a reference to an object:

AtomicReference<String> name = new AtomicReference<>();
name.set("Nacho");
System.out.println("My name is " + name.get());

An important detail is that AtomicReference makes operations on the reference itself atomic. It does not automatically make the internal mutable state of the referenced object thread-safe.

Exercise 1:

Create a project called AtomicCounter from the example shown in section 2 of this document with no synchronization mechanism. You must now use an AtomicInteger attribute (instead of the int attribute of that example), to guarantee that the increment and decrement operations against this object are atomic and thus thread-safe.

2. Accessing arrays

Regarding arrays, Java also provides atomic data types to deal with them. For instance, you can use AtomicIntegerArray to handle integer arrays, or AtomicReferenceArray to handle arrays of object references.

// Create an array of strings with size 10
AtomicReferenceArray<String> names = new AtomicReferenceArray<>(10);

// Add names to some positions
names.set(0, "Arturo");
names.set(1, "Nacho");

// Get names at given positions
System.out.println("Name at 1st position is " + names.get(0));

Atomic array classes provide atomic operations on their individual elements. This does not mean that an arbitrary sequence of operations involving several positions becomes one atomic operation.

2. Accessing arrays

Regarding arrays, Java also provides atomic data types to deal with them. For instance, you can use AtomicIntegerArray to handle integer arrays, or AtomicReferenceArray to handle many other data types.

// Create an array of strings with size 10
AtomicReferenceArray<String> names = new AtomicReferenceArray<String>(10);

// Add names to some positions
names.set(0, "Arturo");
names.set(1, "Nacho");

// Get names at given positions
System.out.println("Name at 1st position is " + names.get(0));

3. Concurrent and synchronized collections

When we want to use data collections in a concurrent program, we have to be very careful with the way we handle these data. Many general-purpose collections, such as ArrayList and HashMap, are not designed for unsynchronized concurrent structural modification.

Java offers both synchronized wrappers in java.util.Collections and dedicated concurrent collections in java.util.concurrent.

3.1. Synchronized collections

Synchronized wrappers can be created with static methods from the Collections class.

// Not synchronized
List<String> list = new ArrayList<>();

// Synchronized wrapper
List<String> syncList = Collections.synchronizedList(list);

The returned wrapper synchronizes its individual operations. It is important that every access to the backing collection is made through the synchronized wrapper; otherwise, the synchronization guarantees can be bypassed.

Iteration is a special case. When traversing a synchronized collection with an Iterator, Spliterator or Stream, we must manually synchronize on the wrapper for the whole traversal:

synchronized (syncList) 
{
    Iterator<String> i = syncList.iterator();

    while (i.hasNext()) 
    {
        System.out.println(i.next());
    }
}

3.2. Concurrent collections

The java.util.concurrent package contains collections specifically designed for concurrent access. They do not all work in the same way, so we should choose them according to the behavior we need.

Blocking collections

A blocking collection can provide operations that wait until an operation can be completed. A common example is LinkedBlockingDeque, which implements BlockingDeque.

It can be created with a fixed capacity:

LinkedBlockingDeque<String> data = new LinkedBlockingDeque<>(10);

data.putLast("One element");
data.putLast("Another element");

String first = data.takeFirst();

If the deque is bounded and full, putFirst / putLast wait until there is room. If it is empty, takeFirst / takeLast wait until an element is available.

Other methods have different behavior. For instance, getFirst does not wait when the deque is empty: it throws an exception instead. Always check the API contract of the operation that you are using.

Blocking queues and deques are particularly useful in producer-consumer designs.

Non-blocking concurrent collections

ConcurrentLinkedDeque is an unbounded thread-safe concurrent deque whose insertion, removal and access operations can safely be performed by multiple threads.

ConcurrentLinkedDeque<String> data = new ConcurrentLinkedDeque<>();

data.addLast("One element");
data.addLast("Another element");

String first = data.pollFirst();

Here, pollFirst() returns null if there is no element available instead of waiting. Methods such as getFirst() or removeFirst() use a different convention and throw an exception if the deque is empty.

Concurrent maps

For key-value data we have, among other choices, ConcurrentHashMap and ConcurrentSkipListMap.

ConcurrentHashMap is a concurrent hash table. ConcurrentSkipListMap, on the other hand, is a scalable concurrent sorted map based on a skip-list data structure.

For example:

ConcurrentSkipListMap<String, String> map = 
    new ConcurrentSkipListMap<>();

map.put("1122", "Ender's game");
map.put("3344", "The Da Vinci Code");

Map.Entry<String, String> element = map.firstEntry();
String isbn = element.getKey();
String title = element.getValue();

System.out.println("First element is " + isbn + " - " + title);

The first entry is determined by the ordering of the keys.

3.3. Synchronized vs concurrent

There is an important difference between synchronized wrappers and dedicated concurrent collections.

For example, ConcurrentHashMap allows concurrent retrievals and a high level of concurrency for updates, whereas ConcurrentLinkedDeque uses a non-blocking concurrent design.

Concurrent collections also commonly provide iterators designed to tolerate concurrent updates. For example, iterators from ConcurrentLinkedDeque and ConcurrentSkipListMap are weakly consistent: they do not throw ConcurrentModificationException merely because the collection changes concurrently, but they are not a fixed snapshot and may or may not reflect concurrent modifications while traversal is taking place.

This behavior is different from the manual external synchronization required when iterating over a Collections.synchronizedList.

3.4. Compound operations and internal objects

A thread-safe collection guarantees the thread-safety properties documented for the collection operations. It does not mean that every arbitrary sequence of operations becomes atomic.

For instance, consider this ConcurrentHashMap:

ConcurrentHashMap<String, Integer> prices = new ConcurrentHashMap<>();
prices.put("Game 1", 20);

Both get and put are thread-safe operations, but this compound sequence is not one atomic operation:

int price = prices.get("Game 1");
prices.put("Game 1", price + 1);

Another thread could modify the same mapping between the get and the put.

For this type of operation, concurrent maps provide atomic compound methods such as compute, computeIfAbsent, computeIfPresent, merge, putIfAbsent and conditional replace.

For example:

prices.compute("Game 1", (title, price) -> price + 1);

For ConcurrentHashMap, the complete compute operation for that key is performed atomically.

There is another important situation: mutable objects stored inside a concurrent collection.

Imagine that we have:

ConcurrentHashMap<String, VideoGame> games = new ConcurrentHashMap<>();
VideoGame game = games.get("Game 1");
game.setPrice(game.getPrice() + 1);

The map safely returns the reference to VideoGame, but the subsequent calls that modify the VideoGame object are operations on that object, not on the map. The concurrent collection does not automatically make VideoGame thread-safe.

If several threads can modify the same VideoGame, its mutable state must also be protected, for example with synchronized methods, a Lock, an appropriate atomic field, or by using immutable objects and atomically replacing the map value.

Exercise 2:

Create a project called ConcurrentVideoGames. In this exercise we are going to see the difference between a thread-safe collection and an atomic compound operation.

Create a ConcurrentHashMap<String, Integer> containing 100 video games. The key will be the game title ("Videogame 1", "Videogame 2", and so on) and the value will be its price as an integer. Set the initial price of every game to 50.

Then create 20 tasks:

In the first version, update every price using separate get and put operations:

int price = games.get(title);
games.put(title, price + 1);   // or price - 1

Launch all the tasks concurrently using an ExecutorService, wait until all of them have finished, and check the final prices. Although the map itself is thread-safe, some prices may be different from the expected value 50 because the read-modify-write sequence is not atomic.

In the second version, reset every price to 50 and replace the previous code with an atomic compound operation:

games.compute(title, (key, price) -> price + 1);

or:

games.compute(title, (key, price) -> price - 1);

Run the program again and verify that every final price is 50.

Finally, explain in a short comment why using a concurrent collection does not automatically make a sequence such as get + calculation + put atomic.