Monitors in java

Monitors are an other mechanism of concurrent programming. It’s a higher level mechanism than semaphores and also more powerful. A monitor is an instance of a class that can be used safely by several threads. All the methods of a monitor are executed with mutual exclusion. So at most one thread can execute a method of the monitor at the same time. This mutual exclusion policy makes easier to work with monitor and to develop the method content of the monitor.

Monitors have an other feature, the possibility to make a thread waiting for a condition. During the wait time, the thread temporarily gives up its exclusive access and must reacquire it after the condition has been met. You can also signal one or more threads that a condition has been met.
There is several advantages on using monitors instead of a lower-level mechanisms :

• All the synchronization code is centralized in one location and the users of this code don’t need to know how it’s implemented.
• The code doesn’t depend on the number of processes, it works for as many processes as you want
• You don’t need to release something like a mutex, so you cannot forget to do it

When we must describe a monitor, we simple use the monitor keyword and describe the methods as common methods :

monitor SimpleMonitor {
    public method void testA(){
        //Some code
    }
 
    public method int testB(){
        return 1;
    }
}

To describe a condition variable, we use the cond keyword. A condition variable is a kind of queue of process who are waiting on the same condition. You have several operations available on a condition, the most important is to signal a process waiting to be awaken and to wait on a condition. There are some similarities between signal/wait operations and P and V of semaphores, but this is a little different. The signal operation does nothing if the queue is empty and the wait operation put always the thread in the waiting queue. The process queue is served in a first come, first served mode. When a thread wakes up after waiting on a condition, it must reacquire the lock before continuing in the code.
Before going further, we must have more informations about the signal operations. When writing monitors, you normally have the choice between several philosophies for the signaling operation :

1. Signal & Continue (SC) : The process who signal keep the mutual exclusion and the signaled will be awaken but need to acquire the mutual exclusion before going.
2. Signal & Wait (SW) : The signaler is blocked and must wait for mutual exclusion to continue and the signaled thread is directly awaken and can start continue its operations.
3. Signal & Urgent Wait (SU) : Like SW but the signaler thread has the guarantee than it would go just after the signaled thread
4. Signal & Exit (SX) : The signaler exits from the method directly after the signal and the signaled thread can start directly. This philosophy is not often used.

The available policies depends on the programming language, in Java, there is only one policy available, the SC one.
In Java there is no keyword to directly create a monitor. To implement a monitor, you must create a new class and use Lock and Condition classes. Lock is the interface is ReentrantLock is the main used implementation, this is the one that we’ll learn to use in the current post. To create a ReentrantLock, you have two constructors, a default constructor and a constructor with a boolean argument indicating if the lock is fair or not. A fair lock indicates that the threads will acquire the locks in the order they ask for. Fairness is a little heavier than default locking strategies, so use it only if you need it. To acquire the lock, you just have to use the method lock and unlock to release it.
The explicit locks have the same memory semantics than the synchronized blocks. So the visibility of the changes is guarantee when you use lock()/unlock() blocks.
So to implement, the monitor example we’ve seen before, we just need to create a class and use the lock to make the mutual exclusion :

public class SimpleMonitor {
    private final Lock lock = new ReentrantLock();
 
    public void testA() {
        lock.lock();
 
        try {
            //Some code
        } finally {
            lock.unlock();
        }
    }
 
    public int testB() {
        lock.lock();
 
        try {
            return 1;
        } finally {
            lock.unlock();
        }
    }
}

The person who’ve already read the other parts of this post set will say that it will be easier to use the synchronized keyword on the two methods. But with synchronized, we will not have the condition variables. If you don’t need condition variables but only locking, it will be easier to use the synchronized blocks instead of Locks.

You can create conditions using the newCondition method on the lock. A condition is a variable of type Condition. You can make the current thread wait on the condition using the await method (and its variant with timeout) and you can signal threads using signal and signalAll methods. The signalAll methods wakes up all the threads waiting on the condition variable.
Let’s try with a simple common example : A bounded buffer. It’s a cyclic buffer with a certain capacity with a start and an end.

import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
 
public class BoundedBuffer {
    private final String[] buffer;
    private final int capacity;
 
    private int front;
    private int rear;
    private int count;
 
    private final Lock lock = new ReentrantLock();
 
    private final Condition notFull = lock.newCondition();
    private final Condition notEmpty = lock.newCondition();
 
    public BoundedBuffer(int capacity) {
        super();
 
        this.capacity = capacity;
 
        buffer = new String[capacity];
    }
 
    public void deposit(String data) throws InterruptedException {
        lock.lock();
 
        try {
            while (count == capacity) {
                notFull.await();
            }
 
            buffer[rear] = data;
            rear = (rear + 1) % capacity;
            count++;
 
            notEmpty.signal();
        } finally {
            lock.unlock();
        }
    }
 
    public String fetch() throws InterruptedException {
        lock.lock();
 
        try {
            while (count == 0) {
                notEmpty.await();
            }
 
            String result = buffer[front];
            front = (front + 1) % capacity;
            count--;
 
            notFull.signal();
 
            return result;
        } finally {
            lock.unlock();
        }
    }
}

So some explications :

1. The two methods are protected with the lock to ensure mutual exclusion
2. Then we use two conditions variables. One to wait for the buffer to be not empty and an other one to wait for the buffer to be not full.
3. You can see that I have wrapped the await operation on a while loop. This is to avoid signal stealers problem that can occurs when using Signal & Continue

And that BoundedBuffer can be easily used with several threads with no problems.
As you can see, you can use monitors to solve a lot of concurrent programming problems and this mechanism is really powerful and performing.
I hope you find this post useful.

Java Locks : Re-entrant locks

Java concurrency library provides more control over synchronization than synchronized. Either we need to control types of access (read and write) separately, or it is cumbersome to use because either there is no obvious mutex or we need to maintain multiple mutexes. So doing it by synchronized will eat lot of time and will be buggy as well.
Thankfully, lock utility classes were added in Java 1.5 and make these problems easier to solve.

Java Reentrant Locks

Java has a few lock implementations in the java.util.concurrent.locks package.
The general classes of locks are nicely laid out as interfaces:

• Lock - the simplest case of a lock which can be acquired and released
• ReadWriteLock - a lock implementation that has both read and write lock types – multiple read locks can be held at a time unless the exclusive write lock is held

Java provides two implementations of these locks that we care about – both of which are reentrant (this just means a thread can reacquire the same lock multiple times without any issue).

• ReentrantLock - as you’d expect, a reentrant Lock implementation
• ReentrantReadWriteLock - a reentrant ReadWriteLock implementation

Extended capabilities with Reentrant locks

The ReentrantLock in the util.concurrent.locks package gives the developers some flexibility here. With ReentrantLock following are some of the options

1. tryLock() : With ReentrantLock, the thread can immediately return if it did not get the lock (if the lock is with some other thread).
2. tryLock(long timeout, TimeUnit unit):With ReentrantLock, the thread can wait for some duration to get hold of the lock. If it does not get the lock within some time, it will return.
3. lockInterruptibly() : With ReentrantLock, the thread waiting for the lock can be interrupted and cause it to come out with InterruptedException.

Now, let’s see some examples. So the general way to use re-entrant lock is like this :

final ReentrantLock _lock = new ReentrantLock();

private void method() throws InterruptedException
{
   //Trying to enter the critical section
   _lock.lock(); // will wait until this thread gets the lock
   try
   {
      // critical section
   }
   finally
   {
      //releasing the lock so that other threads can get notifies
      _lock.unlock(); 
   }       
}

Using optional “fairness” parameter with ReentrantLock
ReentrantLock accepts an optional “fairness” parameter in it’s constructor. Normally what happens is, whenever a thread releases the lock anyone of the waiting threads will get the chance to acquire that lock. But there is no predefined order or priority in the selection of the thread (at least from a programmers perspective).

But if we are specifying the fairness parameter as “true” while creating a new ReentrantLock object, it gives us the guaranty that the longest waiting thread will get the lock next. Sounds pretty nice right?

Use of “Condition” in ReentrantLock
Condition can be considered as a separation of monitor methods (wait(), notify() & notifyAll()). For each ReentrantLock we can define a set of conditions and based on that we can make the threads waiting & things like that.

import java.util.concurrent.locks.Condition;
final Condition _aCondition = _lock.newCondition();
private void method1() throws InterruptedException
{
   _lock.lock();
   try
   {
      while (condition 1)
      {
          // Waiting for the condition to be satisfied
          // Note: At this time, the thread will give up the lock
          // until the condition is satisfied. (Signaled by other threads)
          _aCondition.await();
      }
      // method body
   }
   finally
   {
      _lock.unlock();
   }       

}

private void method2() throws InterruptedException
{
   _lock.lock();
   try
   {
      doSomething();
      if (condition 2)
      {
          // Signaling other threads that the condition is satisfied
          // Wakes up any one of the waiting threads
          _aCondition.signal();

          // Wakes up all threads waiting for this condition
          _aCondition.signalAll();
      }       

      // method body
   }
   finally
   {
      _lock.unlock();
   }
}

Example

We will take the same example of counter again. We have already seen how to implement counter using synchronized keyword here. Here we will see how to implement using Reentrant locks:
Counter.java

package com.vaani.lock;

import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

public class Counter {
    private int count;
    private Lock lock = new ReentrantLock();

    public int getNextValue() {
        try {
            lock.lock();
            count++;
        }
        finally {
            lock.unlock();
            return count;
        }

    }


}

Worker.java
Now worker threads starts on the counter object and start incrementing the value:

package com.vaani.lock;

public class Worker implements Runnable {
    private Counter counter;
    private boolean increment;
    private int count;

    public Worker(Counter counter, boolean increment, int count) {
        this.counter = counter;
        this.increment = increment;
        this.count = count;
    }

    public void run() {
        for (int i = 0; i < this.count; i++) {
            System.out.println(this.counter.getNextValue());


        }
    }
}

Now let's put worker's on work – as in demo:

package com.vaani.lock.demo;
import com.vaani.lock.*;
public class ReentrantLockDemo {
    public static void main(String[] args) throws Exception {
    Counter counter = new Counter();
    Thread t1 = new Thread(new Worker(counter, true, 10000));
    t1.start();
    Thread t2 = new Thread(new Worker(counter, false, 10000));
    t2.start();

    t1.join();
    t2.join();
    System.out.println("Final count: " + counter.getNextValue());
    }
}

Download the source

Source code above can be downloaded from here.

Java locks - Semaphores

Semaphores ensure that only n threads can enter into the critical section of the code at a given time.

Semaphores concept was invented by the famous Dutch computer scientist Edsger Dijkstra. Basically a semaphore is a counter (integer) that allows a thread to get into a critical region if the value of the counter is greater than 0. If it’s the case, the counter is decremented by one otherwise, the thread is waiting until it can go. And when the thread go away from the critical region, the counter is incremented by one to allow one more thread to pass the critical region. A semaphore is created with a certain value for its counter. So, you can execute two actions on a semaphore P and V.

V is also known as signal and it increments the semaphore. P is also known as wait and it decrements semaphore.

By example, if you have a critical that cannot be executed concurrently, you can use a semaphore :

sem mutex = new sem(1)
 
P(mutex)
//Critical region
V(mutex)

So you must always call by yourself the P operation before the critical region and V after it. We call a mutex (mutual exclusion) a semaphore with a value of one. So only one thread can enter the region guarded by the semaphore. This is the most used semaphore. The other use of semaphore is to guard a set of resources like database connections or a data pool. In Java, Semaphore were introduced in jdk5. A semaphore is created using the java.util.concurrent.Semaphore class. You can create easily :

Semaphore mutex = new Semaphore(1);
Semaphore available = new Semaphore(100);

The P and V operations

As discussed above, the P and V operations are represented using the acquire and release methods. The method acquire can be interrupted if the thread is interrupted. There is an ininterruptible version with the method acquireUninterruptibly(). There is also a third version with the tryAcquire method. This method acquire a permit only if there is one permit available, otherwise, this method return false directly. All the waiting methods have also an overloaded version with a timeout. You can also acquire several permits at once using the permits argument to the different versions of acquire methods.

Examples

A little example with a mutex using the same example as the previous post on Java concurrency :

public class Counter {
    private int value = 0;
 
    private final Semaphore mutex = new Semaphore(1)
 
    public int getNextValue() throws InterruptedException {
        try {
            mutex.acquire();
            return value++;
        } finally {
            mutex.release();
        }
    }
}

Consider the test class:

package com.vaani.semaphores.demo;

import java.util.concurrent.Semaphore;

import com.vaani.mutex.Counter;

public class MutexDemo extends Thread{
    public static void main(String[] args) 
     {
         // Limiting No on threads running to 2
        Counter counter = new Counter();
         for(int i=0 ;i<5;i++){
             new MutexThread(String.valueOf(i),counter ).start();
         }
         System.out.println("End of Semaphore Test");
     }
}
 
 
 class MutexThread extends Thread
 {

    private String name  = null;    
     private Counter counter =null;
    public MutexThread(String name,Counter counter_){

         this.name = name;
         this.counter = counter_;
     }
 

     public void run(){
         try {
            System.out.println("Counter is now "+counter.getNextValue());
        } catch (InterruptedException e) {
            // TODO Auto-generated catch block
            e.printStackTrace();
        }
     }
 
 }

For more informations about Semaphore in Java, the best is to consult the Javadoc of the Semaphore class.

Simple Semaphore Example

Consider a simple semaphore in which we just acquire lock and our work and release it:

import java.util.*;
 import java.util.concurrent.*;
class SimpleSemaphoreDemo 
 {
     public static void main(String[] args) 
     {
         // Limiting No on threads running to 2
         Semaphore semaphore = new Semaphore(2);
         for(int i=0 ;i<5;i++){
             new MyThread(String.valueOf(i),semaphore).start();
         }
         System.out.println("End of Semaphore Test");
     }
}
 
 
 class MyThread extends Thread
 {

    private String name  = null;    
     private Semaphore semaphore =null;
    public MyThread(String name,Semaphore semaphore){

         this.name = name;
         this.semaphore = semaphore;
     }
 

     public void run(){
         try{
         semaphore.acquire();
         System.out.println("Thread "+ name +" is start running");
         sleep(500);
         semaphore.release();
         System.out.println("Thread "+ name +" Ends");
         }catch(Exception exp){
             exp.printStackTrace();
         }
     }
 
 }

In this example, we have created 5 threads but no of running threads are limited to only 2, as the line :
Semaphore semaphore = new Semaphore(2); Then we are calling acquire and release functions. The output in this case will look like this :

Thread 0 is start running
End of Semaphore Test
Thread 1 is start running
Thread 0 Ends
Thread 2 is start running
Thread 1 Ends
Thread 3 is start running
Thread 3 Ends
Thread 2 Ends
Thread 4 is start running
Thread 4 Ends

If we put semaphore for 3 threads:
Semaphore semaphore = new Semaphore(3); Now the output will look like this :

Thread 0 is start running
Thread 1 is start running
Thread 2 is start running
End of Semaphore Test
Thread 4 is start running
Thread 2 Ends
Thread 3 is start running
Thread 0 Ends
Thread 1 Ends
Thread 3 Ends
Thread 4 Ends

Conclusion

To conclude, semaphores are a powerful ways to solve concurrency problems, but this is not adapted to all problems. If you need only mutual exclusion, synchronized blocks are a better solutions. The problems with semaphores is that you can forget to call the release method and that can cause deadlock sometimes difficult to find.

Download the source

Source code of this example can be downloaded from here.

Synchronizing with synchronized keyword

Synchronization is a way to make some code thread safe. A code that can be accessed by multiple threads must be made thread safe. Thread Safe describe some code that can be called from multiple threads without corrupting the state of the object or simply doing the thing the code must do in right order.

public class Counter {
    private int value = 0;    
 
    public int getNextValue(){
        return value++;
    }
}

It’s really simple and works well with one thread, but absolutely not with multiple threads. An incrementation like this is not a simple action, but three actions :

Read the current value of “value”
Add one to the current value
Write that new value to “value”
Normally, if you have two threads invoking the getNextValue(), you can think that the first will get 1 and the next will get 2, but it is possible that the two threads get the value 1. Imagine this situation :

Thread 1 : read the value, get 0, add 1, so value = 1
Thread 2 : read the value, get 0, add 1, so value = 1
Thread 1 : write 1 to the field value and return 1
Thread 2 : write 1 to the field value and return 1

These situations come from what we call interleaving. Interleaving describe the possible situations of several threads executing some statements. Only for three operations and two threads, there is a lot of possible interleavings.

So we must made the operations atomic to works with multiple threads. In Java, the first way to make that is to use a lock. All Java objects contains an intrinsic locks, we’ll use that lock to make methods or statement atomic. When a thread has a lock, no other thread can acquire it and must wait for the first thread to release the lock. To acquire the lock, you have to use the synchronized keyword to automatically acquire and release a lock for a code. You can add the synchronized keyword to a method to acquire the lock before invoking the method and release it after the method execution. You can refactor the getNextValue() method using the synchronized keyword :

public class Counter {
    private int value = 0;    
 
    public synchronized int getNextValue(){
        return value++;
    }
}

With that, you have the guarantee that only thread can execute the method at the same time. The used lock is the intrinsic lock of the instance. If the method is static, the used lock is the Class object of Example. If you have two methods with the synchronized keyword, only one method of the two will be executed at the same time because the same lock is used for the two methods. You can also write it using a synchronized block :

public class Counter {
    private int value = 0;
 
    public int getNextValue() {
        synchronized (this) {
            return value++;
        }
    }
}

This is exactly the same as using the synchronized keyword on the method signature. Using synchronized blocks, you can choose the lock to block on. By example, if you don’t want to use the intrinsic lock of the current object but an other object, you can use an other object just as a lock :

public class Counter {
    private int value = 0;
 
    private final Object lock = new Object();
 
    public int getNextValue() {
        synchronized (lock) {
            return value++;
        }
    }
}

The result is the same but has one difference, the lock is internal to the object so no other code can use the lock. With complex classes, it not rare to use several locks to provide thread safety on the class.

There is an other issue with multiple threads : the visibility of the variables. This seems when a change made by a thread is visible by an other thread. For performance improvements, the Java compiler and virtual machines can made some improvements using registers and cache. By default, you have no guarantee that a change made by a thread is visible to an other thread. To make a change visible to an other thread, you must use synchronized blocks to ensure visibility of the change. You must use synchronized blocks for the read and for the write of the shared values. You must make that for every read/write of a value shared between multiple threads.

You can also use the volatile keyword on the field to ensure the visibility of read/write between multiple threads. The volatile keyword ensure only visibility, not atomicity. The synchronized blocks ensure visibility and atomicity. So you can use the volatile keyword on fields that doesn’t need atomicity (if you make only read and write to the field without depending on the current value of the field by example).

You can also note that this simple example can be solved using AtomicInteger, but that will be covered later in an other part of the posts.

Pay attention that trying to solve thread safety on a problem can add new issues of deadlock. By example, if thread A owns the lock 1 and are waiting for the lock 2 and if lock 2 is acquired by thread B who waits on lock 1, there is a deadlock. Your program is dead. So you have to pay great attention to the locks.

There is several rules that we must keep in mind when using locks :

Every mutable fields shared between multiple threads must be guarded with a lock or made volatile, if you only need visibility
Synchronize only the operations that must synchronized, this improve the performances. But don’t synchronize too few operations. Try to keep the lock only for short operations.
Always know which locks are acquired and when there are acquired and by which thread
An immutable object is always thread safe
Here we are, I hope that this post helps you to understand thread safety and how to achieve it using intrinsic locks. In the next posts, we’ll see another synchronization methods.

How to synchronize a static variable of a class ?

There are some ways(3 to my knowledge, but may be more), by which a static variable can be synchronized in java.

1) Use a synchronized static method. This synchronizes on the class object.

public class Counter {
    private static int count = 0;

    public static synchronized void incrementCount() {
        count++;
    }
} 

2) Explicitly synchronize on the class object, using synchronize on ClassName.class

public class Counter {
    private static int count = 0;

    public void incrementCount() {
        synchronize (Test.class) {
            count++;
        }
    }
} 

3) Synchronize on some other static object.

public class Counter {
    private static int count = 0;
    private static final Object countLockHelper = new Object();

    public void incrementCount() {
        synchronize (countLockHelper) {
            count++;
        }
    }
} 

Method 3 is best in many cases because the lock object is not exposed outside of your class. So if you create instance of these class, they will be synchronized on the same static object.

But if you just using some basic type like integer here in case of counter, consider using an AtomicInteger or another suitable class from the java.util.concurrent.atomic package:

public class Counter {

    private final static AtomicInteger count = new AtomicInteger(0); 

    public void incrementCount() {  
        count.incrementAndGet();
    }  
}

Why are thread related methods present in Object class

The threading methods are defined in Object class because these methods are related to object locks and any object can act as a lock and hence these methods are defined in the Object class. These methods are wait(), notify() and notifyAll(). These methods are final and return nothing. All of them must be called from a synchronized block or method and the thread calling them should pocess the lock on the object.
While calling wait() method, the thread to release the lock on Object on which wait() method is called and on calling the notify()/notifyAll() makes the other capture the lock.

Class level lock vs instance level lock

A class level lock is the lock which makes all object of a class to wait until the corresponding lock is not released.
e.g

Class A{
   static synchronized void foo(){}
}

Here the method foo is synchronized and hence all the threads on all the objects of the class will wait until the object currently running the foo method completes its execution.

Similarly an instance level lock makes all the threads started using the instance of the class to wait until the lock is not released.
e.g.

Class A{
    //static is missing
    synchronized void bar(){}
}

Here all the threads started from the object which is currently executing the bar method will wait until the current threads completes its execution. Note that other threads of other objects can execute the bar method while another object's thread is executing the bar method.

Locking an Object

The code segments within a program that access the same object from separate, concurrent threads are called critical sections. In the Java language, a critical section can be a block or a method and are identified with the synchronized keyword. The Java platform then associates a lock with every object that has synchronized code.

Sample Code
In the producer/consumer example, the put and get methods of the CubbyHole are the critical sections. The Consumer should not access the CubbyHole when the Producer is changing it, and the Producer should not modify it when the Consumer is getting the value. So put and get in the CubbyHole class should be marked with the synchronized keyword.
Here's a code skeleton for the CubbyHole class:

public class CubbyHole {
private int contents;
private boolean available = false;

public synchronized int get() {
...
}

public synchronized void put(int value) {
...
}
}

Note that the method declarations for both put and get contain the synchronized keyword. Hence, the system associates a unique lock with every instance of CubbyHole (including the one shared by the Producer and the Consumer). Whenever control enters a synchronized method, the thread that called the method locks the object whose method has been called. Other threads cannot call a synchronized method on the same object until the object is unlocked. So, when the Producer calls CubbyHole's put method, it locks the CubbyHole, thereby preventing the Consumer from calling the CubbyHole's get method:

public synchronized void put(int value) {
// CubbyHole locked by the Producer
..
// CubbyHole unlocked by the Producer
}

When the put method returns, the Producer unlocks the CubbyHole. Similarly, when the Consumer calls CubbyHole's get method, it locks the CubbyHole, thereby preventing the Producer from calling put:

public synchronized int get() {
// CubbyHole locked by the Consumer
...
// CubbyHole unlocked by the Consumer
}

The acquisition and release of a lock is done automatically and atomically by the Java runtime system. This ensures that race conditions cannot occur in the underlying implementation of the threads, thus ensuring data integrity. Synchronization isn't the whole story. The two threads must also be able to notify one another when they've done their job. Learn more about that after a brief foray into reentrant locks.

See also class level lock vs instance level lock.