Latch API in Java : CountDownLatch

Literally, latch means a device for keeping a door or gate closed.  Its meaning is analogous to a gate in Java as well. So if latch (gate) is open, everyone can pass through it but when it's shut, no one is allowed to cross over. With this as background let's go in detail:

It's one of the advanced threading/concurrency concepts of Java. Java provides a Latch API named as  CountDownLatch which got introduced in Java 5 ( java.util.concurrent package). So now on; I will refer Latch and CountDownLatch interchangeably to refer to the same thing. Latch is a synchronizer that can delay the progress of threads until it reaches its terminal state. [A synchronizer is any object that coordinates the control flow of threads based on its state] So it is used to synchronize one or more tasks by forcing them to wait for the completion of a set of operation being performed by other tasks.

CountDownLatch in action

Let me start directly with a simple example to stress on the fundamentals of this API. Below sample program has two tasks (as taskone() and tasktwo()) represented as methods. And I want to make sure that taskone() should get completed before tasktwo(). 

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

public class CountDownLatchTest {
 static CountDownLatch latch;

 CountDownLatchTest(final int count) {
  latch = new CountDownLatch(count);
 }

 public void firstTask() {
  Runnable s1 = new Runnable() {
   public void run() {
    try {
         System.out.println("waiting....");
         TimeUnit.SECONDS.sleep(5);
    } catch (InterruptedException e) {
         e.printStackTrace();
    }
    // finish first activity before last line
    latch.countDown();
   }
  };
  Thread t = new Thread(s1);
  t.start();
 }

 public void secondTask() {
  Runnable s2 = new Runnable() {
   public void run() {
    try {
         System.out.println("wait....");
         latch.await();
         // perform task here
         System.out.println("after wait.... done");
    } catch (InterruptedException e1) {
         e1.printStackTrace();
    }
   }
  };
  Thread t = new Thread(s2);
  t.start();
 }

 public static void main(String[] args) throws InterruptedException {
  CountDownLatchTest cdlt = new CountDownLatchTest(1);
  cdlt.secondTask();
  TimeUnit.SECONDS.sleep(5);
  cdlt.firstTask();
 }
}

 Output:

 wait....
waiting....
after wait.... done

Above example shows CountDownLatch attribute getting initialized to a value of 1 through constructor. Two tasks in above example are synchronized though CountDownLatch. Task2 i.e. secondTask() should wait for completion of Task1 i.e. firstTask(). Run above example and notice the sequence in which output appears on the console. 

Please note few important points

1. Any task that calls await() on the object will block until the count reaches zero or it's interrupted by another thread. secondTask() gets blocked after call to await(); evident from output.
2. Call countDown() on the object to reduce the count. The task that call countDown() are not blocked.  This cal signals the end of the task. This method need to be called at the end of the task.
3. As soon as count reaches zero; threads awaiting starts running. 
4. The value of count which is passed during creation of latch object is very important. It should be same as the number of task which needs to be finished first. If count is 5 then first task should be called five times to make sure that count has reduced to 0.
5. You can also use wait and notify mechanism of Java to achieve the same behavior but code will become quite complicated. 
6. One of the disadvantage of CountDownLatch is that its not reusable once count reaches to zero. But Java provides another concurrency API called CyclicBarrier for such cases. 

Usage of CountDownLatch

1. Use this when your current executing thread/main thread needs to wait for the completion of other dependent activities. 
2. Ensure that a service doesn't start until other services on which it depends have not completed.
3. In a multi-player game like RoadRash; wait for all players to get ready to start the race. 

---
do post your comments/questions !!!

Semaphore in Java

Let's start with below problems; before starting on Semaphore.

1. Implement a Database connection pool which will block if no more connection is available instead of failing and handover connection as soon as it's available
2. Implement a thread pool
3. Create a blocking bounded collection
4. Limiting number of http connection to an external site

Note : Bounded puts an upper limit on the size of collection. 

And block(ing) here means that, wait for the collection to become non-empty when retrieving an element, and wait for space to become available when storing an element. 

(One of the) Solution to all above problems is, Semaphore!
Recall that, in normal monitor lock (implicit lock), only one thread can access a resource. So, if you want more than one thread to access a resource at the same time, Semaphore comes into play. Semaphores are used to restrict number of threads which can access a resource or perform a given action at the same time. To achieve this, semaphore maintains a counter which keeps track of the number of resources available.

When a thread requests access to resource, semaphore checks the variable count, and if it's less than total count, then grants access and subsequently reduces the available count. If count is equal to maximum allowed count then it asks thread to wait. If resource count is one (0/1 or on/off) then its called as Binary Semaphore, otherwise its called as counting semaphore.

Semaphore got introduced into standard Java library in version 5.

Limit Number of Http Connections

Below example show how can you limit the number of http connection in your system.

import java.io.IOException;
import java.net.URL;
import java.net.URLConnection;
import java.util.concurrent.Semaphore;

/**
 * Limit Maximum number of URL connection allowed through Counting Semaphore
 * @author Sid
 *
 */
class UrlConnectionManager {
 private final Semaphore semaphore;
 private final int DEFAULT_ALLOWEED = 10;

 UrlConnectionManager(int maxConcurrentRequests) {
  semaphore = new Semaphore(maxConcurrentRequests);
 }
 
 UrlConnectionManager() {
  semaphore = new Semaphore(DEFAULT_ALLOWEED);
 }

 public URLConnection acquire(URL url) throws InterruptedException,
   IOException {
  semaphore.acquire();
  return url.openConnection();

 }

 public void release(URLConnection conn) {
  try {
   // clean up activity if required
  } finally {
   semaphore.release();
  }
 }
}

In above example; semphore maintains a set of permits or pass. Method, semaphore.acquire(), blocks if necessary until permit is available and then takes it. And each semaphore.release() adds/returns a permit. Semaphore is like a gatekeeper which keeps track of number of visitors allowed in a building.

Blocking Bounded LinkedList

You can also use semaphore to turn any collection into blocking bounded collection. Let's see how to do the same for LinkedList. Assume that, only add and remove operations are allowed. Below class provides main method to test it. Note that, at most 5 add operations are allowed.

import java.util.Collections;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * Minimilistic Bounded List  
 * @author Sid
 *
 */
public class BoundedLinkedList<T> {
 private List<T> list;
 private Semaphore semaphore;

 public BoundedLinkedList(final int bound) {
  list = Collections.synchronizedList(new LinkedList<T>());
  semaphore = new Semaphore(bound);
 }

 public void add(T o) throws InterruptedException {
  semaphore.acquire();
  list.add(o);
  System.out.println(" Added element :"+ o);
 }
 
 public void remove(T o) {
  if (list.contains(o)) {
   list.remove(o);
   semaphore.release();
   System.out.println(" Removed element :"+ o);
  }
 }
 
 /**
  * Test Method
  * Adds 6 elements; then waits for 5 seconds; removes one element
  */
 public static void main(String[] args){
     final BoundedLinkedList<Integer> bll = new BoundedLinkedList<>(5);
  new Thread(new Runnable(){
   public void run(){
    for(int i = 1; i <= 6; i++)
     try {
      bll.add(i);
     } catch (InterruptedException e) {
      e.printStackTrace();
     }
   }
  }).start();
  
  try {
   TimeUnit.SECONDS.sleep(5);
  } catch (InterruptedException e) {
   e.printStackTrace();
  } 
  
  bll.remove(3);
 }

}

Output
 Added element :1
 Added element :2
 Added element :3
 Added element :4
 Added element :5
 Removed element :3
 Added element :6

Important points to remember

1. You should be very careful to make sure that you are releasing after acquire. You can miss it due to programming error or any exception. 
2. Long held lock can cause starvation 
3. Method, release doesn't have to be called by the same thread which called acquire. This is an important property that we don't have with normal mutex in Java.
4. You can increase number of permits at runtime (you should be careful though). This is because number of permits in a semaphore is not fixed, and call to release will always increase the number of permits, even if no corresponding acquire call was made. 

---
keep coding !!!