The Cost of Concurrency

Concurrency is not free!

Modern libraries provide a wonderful abstraction for programmers, so doing certain task concurrently or asynchronously is quite trivial. It is as simple as instantiating an object and calling few methods on it, and you are done! These libraries are abstracted in such a way that they don't even remind to programmers that you are going to deal with threads. And this is where the lazy programmer can take things for granted.

You need to process 100 task, create 50 threads.
     Collection<Task> task = fetchTasks();   //from somewhere
     int numberOfThreads = 50;
    obj.executeConcurrently(tasks, numberOfThreads);
In object oriented world, all it takes is a method call. 

To understand the cost of concurrency, let's take a step back and ask yourself how is it implemented? It is implemented through locks. Locks provide mutual exclusion and ensure that the visibility of change occurs in an ordered manner. 

Locks are expensive because they require arbitration when contended. This arbitration is achieved by a context switch at the OS level which will suspend threads waiting for lock until it is released. Context switch might cause performance penalty as OS might decide to do some other housekeeping job and so will lose the cached instruction and data. This is even more evident in multicore CPUs where each core has its own cache.  In the worst case, this might cause latency equivalent to that of an I/O operation. 

Another aspect of concurrency is managing the lifecycle of threads. OS does the dirty job of creating threads and managing them on behalf of your platform (or runtime environment). There are certain limits on the number of threads which can be created at the system level. So definitely, proper thoughts should be given on how many threads are required to accomplish a job.

Don't blindly decide to execute task concurrently!

Concurrency or Thread Model of Java

Thread Model in Java is built around shared memory and Locking!

Concurrency basically means two or more tasks happen in parallel and they compete to access a resource.  In the object-oriented world, the resource would be an object which could abstract a database, file, socket, network connection etc. In the concurrent environment, multiple threads try to get hold of the same resource. Locks are used to ensuring consistency in case there is a possibility of concurrent execution of a piece of code. Let's cover these aspects briefly:

Concurrent Execution is about

1. Mutual Exclusion or Mutex
2. Memory Consistency or Visibility of change

Mutual Exclusion

Mutual exclusion ensures that at a given point of time only one thread can modify a resource.  If you write an algorithm which guarantees that a given resource can be modified by (only) one thread then mutual exclusion is not required.  

Visibility of Change

This is about propagating changes to all threads. If a thread modifies the value of a resource and then (right after that) another thread wants to read the same then the thread model should ensure that read thread/task gets the updated value. 

The most costly operation in a concurrent environment contends write access. Write access to a resource, by multiple threads, requires expensive and complex coordination. Both read as well as write requires that all changes are made visible to other threads. 

Locks

Locks provide mutual exclusion and ensure that visibility of change is guaranteed (Java implements locks using the synchronized keyword which can be applied on a code block or method).

Read about the cost of locks, here. 

Related Post: http://geekrai.blogspot.in/2014/02/atomic-operations-in-java.html

Case Study: http://martinfowler.com/articles/lmax.html
References: Java Memory Model

Implementing Thread Pool in Java

In the previous post, I discussed Thread Pool fundamentals. I strongly believe such confusing concepts should be best explained through code; so here goes this dedicated post!

This post has a prototype thread pool implementation in Java. I have abstracted Runnable into a Task and Thread into worker to avoid the confusion. It also serves the purpose of clearly differentiating task and thread.

Implementation

Classes can be classified broadly in below two categories.
Job Queue: Task, TaskQueue, and TaskQueueImpl
Worker/Thread Pool: Worker and WorkerPool

Job queue component is about tasks and their storage in the queue. Worker pool manages lifecycle of threads/workers.

package pool.thread;

/**
 * Abstracts task; Notice it implements Runnable.
 * It finds all prime numbers between two values (i.e. start and end)
 * 
 */
public class Task implements Runnable {
 private static int counter = 1;
 private String taskId;
 private int start;
 private int end;

 public Task(int start, int end) {
  this.taskId = "task-" + counter++;
  this.start = start;
  this.end = end;
 }

 @Override
 public void run() {
  System.out.println(" \n Prime Numbers in range [" + this.start + "-"
    + this.end + "]");
  while (start <= end) {
   if (isPrime(start)) {
    System.out.print(" " + start);
   }
   start++;
  }
 }

 private boolean isPrime(int num) {
  if (num > 1 && num % 2 == 0) {
   return false;
  }
  int sqrRoot = (int) Math.sqrt(num);
  for (int i = 3; i < sqrRoot; i += 2) {
   if (num % i == 0) {
    return false;
   }
  }
  return true;
 }

 public String toString() {
  return "id:" + this.taskId;
 }
}

package pool.thread;

/**
 * Interfaces supported by Task queue
 */
public interface TaskQueue {
 public Task getNextJob();
 public void addJob(Task task);
 public boolean isEmpty();
}

package pool.thread;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

/**
 * Provides TaskQueue implementation
 * It uses BlockingQueue to implement queue
 */
public class TaskQueueImpl implements TaskQueue {
 private BlockingQueue<Task> queue;

 public TaskQueueImpl(int size) {
  queue = new ArrayBlockingQueue<Task>(size);
 }

 public Task getNextJob() {
  return queue.poll();
 }

 @Override
 public void addJob(Task task) {
  try {
   queue.put(task);
  } catch (InterruptedException e) {
   e.printStackTrace();
  }
 }

 @Override
 public boolean isEmpty() {
  return queue.isEmpty();
 }
}

package pool.thread;

/**
 * Abstracts the workers/threads of the thread pool
 * Once the queue is empty; the worker kills itself
 */
public class Worker extends Thread {
 private volatile boolean isStopped;
 private TaskQueue jobQueue;
 private String workerId;
 private static int counter = 1;

 public Worker(TaskQueue jobQueue) {
  isStopped = false;
  this.jobQueue = jobQueue;
  this.workerId = "worker-" + counter++;
 }

 public void run() {
  while (!isStopped) {
   System.out.println();
   if (!this.jobQueue.isEmpty()) {
    Task task = (Task) this.jobQueue.getNextJob();
    task.run();
    System.out.println("\n Above Task, " + task.toString()
      + "; finished by worker, " + this.toString());
   }else{
     this.kill();
   }
  }
 }

 public void kill() {
  this.isStopped = true;
 }

 public void startWorker() {
  this.start();
 }

 public String toString() {
  return "id :" + this.workerId;
 }

}

package pool.thread;

import java.util.ArrayList;
import java.util.List;

/**
 * Thread pool / worker pool implementation
 * Note that - methods are synchronized to make this class thread safe
 */
public class WorkerPool {
 private List<Worker> pool;
 private int maxThreads;

 public WorkerPool(int maxThreads) {
  this.maxThreads = maxThreads;
  this.pool = new ArrayList<>(maxThreads);
 }

 public synchronized void execute(TaskQueue jq) {
  for (int i = 0; i < maxThreads; i++) {
   Worker worker = new Worker(jq);
   pool.add(worker);
   worker.startWorker();
  }
 }

 public synchronized void stop() {
  while(!pool.isEmpty()){
   for (Worker worker : pool) {
    worker.kill();
   }   
  }
 }
}

Client Code
        TaskQueue jq = new TaskQueueImpl(5);
        jq.addJob(new Task(1, 100));
        jq.addJob(new Task(101, 200));
        jq.addJob(new Task(201, 300));
        jq.addJob(new Task(301, 400));
        jq.addJob(new Task(401, 500));
 
        WorkerPool wp = new WorkerPool(2);
        wp.execute(jq);

Output
Prime Numbers in range [1-100]
 1 3 5 7 9 11 13 15 17 19 23 25 29 31 35 37 41 43 47 49 53 59 61 67 71 73 79 83 89 97
 Above Task, id:task-1; finished by worker, id :worker-2


 Prime Numbers in range [101-200]
 101 103 107 109 113 121 127 131 137 139 143 149 151 157 163 167 169 173 179 181 191 193 197 199
 Above Task, id:task-2; finished by worker, id :worker-2


 Prime Numbers in range [201-300]
 211 223 227 229 233 239 241 251 257 263 269 271 277 281 283 289 293
 Above Task, id:task-3; finished by worker, id :worker-2


 Prime Numbers in range [301-400]
 307 311 313 317 323 331 337 347 349 353 359 361 367 373 379 383 389 397
 Above Task, id:task-4; finished by worker, id :worker-2


 Prime Numbers in range [401-500]
 401 409 419 421 431 433 439 443 449 457 461 463 467 479 487 491 499
 Above Task, id:task-5; finished by worker, id :worker-2

---
keep coding !!!
do post your feedback; 
this post marks the end of a fruitful year, 2014 :)

Thread pool in Java

Thread Pool, literary means pool of thread. But, the literal  meaning is a bit confusing, as just having a pool of threads is hardly of any value. So, just remember that thread pool is not just another Object pool (like String pool). I will be talking about it in detail here!

The missing story from the literal meaning of the thread pool is that it also uses a queue to manage tasks. 

The pool of threads + Job Queue makes story complete, and hence just calling it thread pool is bit confusing and even misleading. This term is not specific to Java and used in DataBases and other programming languages as well. It's not easy to change its intended meaning or rename it to make it clearer (it existed even before Java was born and definitely going to last beyond Java!).

The fundamental idea behind an object pool is that creation and destruction (finalization or garbage collection) of any resource (read objects here, and even threads) is dearer. So, if you have a pool of readily available resources, they can be reused.

 

Task vs Thread

Most of the programmers use task and thread interchangeably and mean the same thing; that's plain WRONG! A thread executes a task.

Tasks are independent activities and don't depend on the state or side effect of other tasks. This independent nature of tasks facilitates concurrency and hence can be executed by multiple threads at the same time. Also, tasks are usually finite; they have a clear starting point and they eventually terminate. Tasks are logical unit of work, and threads are a mechanism by which tasks can be run asynchronously.

Runnable is the default task abstraction in Java. Don't get confused by trying to understand tasks and threads the way we create threads in Java; by either extending Thread class or implementing Runnable.

          Runnable task = new Runnable(){
            public void run(){
                System.out.println("inside task");
            }
        };  //end of task definition 

      
    Thread thread = new Thread(task);  //note that task goes as argument
        thread.start(); 

Above example helps to understand difference between task and thread. It clearly explains that thread takes task as argument to process/execute it. Calling start method on thread will internally call the run method declared in Runnable (of course in a separate thread). Other task abstraction in Java is Callable which got added in J2SE 5.0.

On similar lines (of task and thread), thread pool has a task queue as well.

Thread Pool

Thread Pool manages a homogeneous pool of workers or threads and it is bound to a work queue holding tasks. Workers from the pool go and pick the next task from the queue and execute them independently. Once a task is finished, the workers fetch the next task from the queue if it has one, otherwise, workers keep on waiting for next task in the queue.

Executor framework (link), added in Java 5.0 provides static factory methods in Executors to create thread pool (like newFixedThreadPool(), newCachedThreadPool(), newSingleThreadExecutor(), newScheduledThreadPool()). The Executor framework uses Runnable as its basic task representation. If you want more control, Callable is a better choice.

Let's zoom into the interfaces provided by two major components of ThreadPool.

Task Queue: Used for holding tasks in a queue so that workers can pick and execute them. The queue can be implemented as a FIFO or Priority Queue. At least, below methods should be supported by a task queue.

public interface TaskQueue {
    public Task getNextJob();
    public void addJob(Task task);
    public boolean isEmpty();
}

Worker Pool: Used for holding threads. The pool should provide an interface to execute task submitted to it. So the execute method takes tasks queue as an argument.

public interface WorkerPool {
    public void execute(TaskQueue queue);
}

Time to wind up on the fundamentals.
Next post (link), I have covered thread pool implementation in Java.

wait and notify in Java

The lock/monitor mechanism provided in Java prevents more than one thread from accessing a shared resource. Java achieves it by allowing only one thread to get hold of methods/block using synchronization technique. Thread safety comes at the cost of no direct communication between threads.

Some problem require inter-thread communication, though; like producer-consumer problems. Producers and consumers co-operate with each other and decide next action, which is difficult to achieve using default monitor lock. There is a need to bring in coordination between multiple threads. Java provides wait/notify methods to handle such problem which require two or more threads/tasks to talk to each other.

wait and notify methods

These methods are part of the superclass, Object. Let's cover them briefly before jumping to the example:

wait/wait(..): This method provides a way to synchronize activities between tasks. Within a task when you call wait(), the execution of the task gets suspended and the lock on the object is released. Once the lock is released, it can get acquired by another task.
Overloaded wait method (i.e. wait(100) takes an argument in milliseconds (similar to sleep). 

notify/notifyAll: Obvious question after call to wait is, what makes the task to resume? 

Task resumes if the same object calls notify or notifyAll method. If more than one tasks are in wait state, then notifyAll needs to be called instead of notify. So notify wakes up task(s) which is/are in wait state.

Example to understand wait and notify

Let's take a trivial example to show co-operation between two threads using wait and notify methods. ThreadCoordination class has two methods, one generates the random number and other one prints the same (getNextRandom() method waits until the next number is generated through the method, generateNextRandom()). The main method, test these two methods by creating two threads. These threads are generating and printing first ten random numbers.

 import java.util.Random;  
 import java.util.concurrent.TimeUnit;  
 public class ThreadCoordination {  
      private volatile int randomNum = 0;  
      Random rand = new Random(101);  
      public synchronized void generateNextRandom() throws InterruptedException{  
           while(randomNum != 0){  
                wait();  
           }  
           TimeUnit.SECONDS.sleep(2); //delay of 2 seconds  
           randomNum = rand.nextInt(101);  
           notify();  
      }  
      public synchronized void getNextRandom() throws InterruptedException {  
           while(randomNum == 0){  
                wait();  
           }  
           System.out.println(" Next Number is : "+ randomNum);  
           randomNum = 0 ; //reset the value  
           notify();  
      }  
      //Test method  
      public static void main(String[] args) {  
           final ThreadCoordination tc = new ThreadCoordination();  
           new Thread(){  
                public void run(){  
                     for(int i=0; i<10; i++){  
                          try {  
                               tc.getNextRandom();  
                          } catch (InterruptedException e) {  
                               e.printStackTrace();  
                          }  
                     }  
                }  
           }.start();  
           new Thread(){  
                public void run(){  
                     for(int i=0; i<10; i++){  
                          try {  
                               tc.generateNextRandom();  
                          } catch (InterruptedException e) {  
                               e.printStackTrace();  
                          }  
                     }  
                }  
           }.start();  
      }  
 }  

Output:
 Next Number is :21
 Next Number is :27
 Next Number is :82
 Next Number is :60
 Next Number is :69
 Next Number is :13
 Next Number is :11
 Next Number is :43
 Next Number is :17
 Next Number is :56

Note, both methods are using wait as well as notify to complement each other.  getNextRandom() waits until next random number gets generated in method generateNextRandom(). And generateNextRandom() waits until the generated number gets printed (and variable gets resets).
Also, wait() needs to be called from the loop; calling it from an if condition will not help. 

BlockingQueue Example

Let's take another classic example of coordination between two different tasks. Below is a simplified blocking queue example.

 import java.util.LinkedList;  
 import java.util.Queue;  
   
 public class SimpleBlockingQueue<T> {  
      private Queue<T> queue = new LinkedList<T>();  
      private int size;  
   
      public SimpleBlockingQueue(int size) {  
           this.size = size;  
      }  
      public synchronized void put(T element) throws InterruptedException {  
           while (queue.size() == size) {  
                wait();  
           }  
           queue.add(element);  
           notify();  
      }  
      public synchronized T take() throws InterruptedException {  
           while (queue.isEmpty()) {  
                wait();  
           }  
           T item = queue.remove();  
           notify();  
           return item;  
      }  
 }  

Note, Java provides a BlockingQueue API; prefer it instead of re-inventing the wheel. 

 

Some Important Points 

why wait and notify are part of the base class?

Sounds more like thread concepts but put in superclass, Object. This is required because these methods get called from the synchronized block/ method and they manipulate locks. And these locks are objects (in Java). 

sleep vs wait

The lock doesn't get released during the call of sleep but that's not the case with wait. If wait is called without any parameter, it can come out due to notify() or notifyAll(). If wait is called with parameter then it can respond to time out as well as notify()/notifyAll().

Also, wait is always called from synchronized block/method but sleep can be called within non-synchronized methods as well. 

what if wait/notify is called from non-synchronized block/method

The program will compile. But you will get a runtime exception, IllegalMonitorStateException. This basically means that task calling wait(), notify(), or notifyAll() MUST own the lock for the object.

prefer API provided by the language
Instead of writing code using wait/notify mechanism, prefer inbuilt Java API. 

---
keep coding !!!

Interrupting a Thread in Java

I have been working on a priority change request which was supposed to be fixed by end of business hour today. Today morning, I was totally engrossed into it when my boss interrupted me to convey that CR has been deferred. Gosh! what a relief it was. 

An interrupt is basically an indication to a thread that it should do something else instead of whatever it is doing right now. Something else is subjective here;  thread could abort or terminate the task, ignore the interrupt request after acknowledging it or it could even ignore it altogether as if nothing happened.
Usually interrupt mechanism works between two threads. One thread raises interrupt for other thread. Once interrupt is received by the other thread, it can act as per its wish. Thread can also interrupt itself.

Interrupting a Thread

Interrupting a thread is quite trivial. Just get a handle to thread and call interrupt method on it. Done!

              t.interrupt();

Below class demonstrates how to call interrupt on a thread. Main thread creates a new thread which keeps on incrementing and printing a counter.

 public class InterruptDemo1 {   
    static int counter = 0 ;   
    public static void main(String[] args) {   
       Runnable task = new Runnable(){   
         public void run(){   
            while(true){   
              System.out.println(" val :"+ counter++);                
            }   
         }   
       };   
       Thread secondThread = new Thread(task);   
       secondThread.start();     
       secondThread.interrupt();      
    }   
  }   

If you run above program, it will go on and on, printing the incremented value (couldn't see it stopping for 10 minutes on my machine).

In the last line, main thread interrupts secondThread. But it is not making any difference to the execution of the program. Is it all cool?

Handling Interrupt

As far as raising interrupt is concerned, above class demonstrates it perfectly. The problem is actually with second thread which is not handling the raised interrupt. And that's the reason why you see uninterrupted value on your console. So even if an interrupt is raised the thread can chose to ignore it.

In modified class below, interrupt gets handled by the task/thread.

 public class InterruptDemo2 {  
      static int counter = 0 ;  
      public static void main(String[] args) {  
           Runnable task = new Runnable(){  
                public void run(){  
                     while(true){  
                          System.out.println(" val :"+ counter++);  
                          if(Thread.interrupted())  //Thread refers to current thread
                               return;  
                     }  
                }  
           };  
           Thread secondThread = new Thread(task);  
           secondThread.start();  
           secondThread.interrupt();  
      }  
 }  

Now interrupt is getting handled in the second thread (run above program to confirm the same).

Details on Interrupt methods of Thread

 package java.lang;  
 public class Thread implements Runnable{  
      public void interrupt(){..}  
      public boolean isInterrupted(){..}  
      public static boolean interrupted(){..}  
   ...  
 }  

As discussed in previous examples, interrupt() method can be used to interrupt a thread. Thread maintains this state through a boolean interrupted status flag. This flag can be used to check the status of interruption by calling isInterrupted() method. And interrupted() method returns the status as well as clears the value. This is the only way to clear the interrupt status flag (i.e. sets it to false). Just be careful in the usage of this method though. Your task/thread might eat the interrupt. Best approach is, either handle interruption ( by exiting or throwing InterruptedException) or restore the interruption status by calling interrupt() again.

Blocking methods like Thread.sleep and Object.wait can also detect interrupt on a thread. These methods respond to interrupt by clearing the status flag and also by throwing InterruptedException.

Stopping a Thread

Stopping a task (and thread) safely and quickly is not easy. We need a cooperative mechanism to make sure that stopping the thread doesn't leave it in an inconsistent state. This was the reason why Java designers deprecated stop and suspend method from Thread class.

One of the most wildly used approach is, keep a thread-safe flag to notify to the thread that it can cancel the task. But this is not a reliable approach, task might not check the cancelled flag if it's making blocking API call.

          volatile boolean cancelled = false;    //class attribute
           //task component
            if(!cancelled){
                  //blocking api call
            }else{
               return;
            }

Interruption mechanism discussed in this post, is safest and most sensible approach to stop a thread. So you can design your task in such a way that when it receives interrupt, it can come out of task gracefully.

--happy interruption!

Callable and Future in Java

Runnable is default abstraction for creating a task in Java. It has a single method run() that accepts no arguments and returns no value, nor it can throw any checked exception. To overcome these limitations, Java 5 introduced a new task abstraction through Callable [javadoc ] interface.

public interface Callable<V>{
     V call() throws Exception;
}

Evident from the Callable interface shown above, call() method can return an Object, or more specifically any generic type. Like Runnable, Callable is also designed to be executed by threads. But you can't directly pass a Callable into Thread for execution. Check below snippet :

         Callable task = new Callable(){
  public Integer call(){
System.out.println("inside call method");
   return 1;
}
        };
Thread t = new Thread(task);    //doesn't compile

The last line in above code will not compile. It gives below compilation error message:  
"The constructor Thread(Callable) is undefined".
It tells Thread can't take a Callable argument.

Recall that Thread class implements Runnable interface, and it has been with Java since version 1.0, whereas Callable got added to the language in version 5.0 (or 1.5) only.

Running a Callable Task

Passing Callable into Thread for execution is out of question. Java SE 5 provides ExecutorService to execute the Callable object. The service accepts a Callable object.

        <T> Future<T> submit(Callable<T> task);

As above definition shows, submitting a Callable object to ExecutorService returns Future object. Future represents the life cycle of a task and is discussed below:

        Callable<List<T>> task = new Callable<List<T>>(){
               public List<T> call(){
                    List<T> results = new ArrayList<T>();
                   
                    //computation
                   
                    return results;
               }
        };

       ExecutorService es = Executors.newSingleThreadExecutor();
       Future<List<T>> future = es.submit(task);
       List<T> data = future.get();
       System.out.println("List Data :"+ data); 

The get() method of Future blocks, until the task is completed. This is equivalent to join() (i.e. t.join()) in normal thread case.

A closer look at Future

Future represents the life cycle of a task and provides methods to test whether the task has completed or been canceled, retrieve the result, and cancel the task. The behavior of get method depends on the state of the task (yet to start, completed, running). Get method returns immediately or throws Exception if the task has already completed. If the task is not completed then it blocks until the task completes. 

Future class methods

Straight from Java Doc :

Future represents the result of an asynchronous computation. Methods are provided to check if the computation is complete, to wait for its completion, and to retrieve the results of the computation.  The results can only be retrieved using method get when the computation has completed, blocking if necessary until it is ready. Cancellation is performed by the cancel method. Additional methods are provided to determine if the task completed normally or was canceled. Once a computation has completed, the computation cannot be canceled. If you would use a Future for the sake of cancellability but not provide a usable result, you can declare the type of the form Future<?> and return null as a result of the underlying task.

Related Post : Executor Framework in Java

----
post your comments/questions below !!!