Java Thread - print Odd Even Numbers

This is possible by :  EvenOddPrint [synchronization + wait () + notifyAll () ]+ 2 Threads (odd/even) + Task [ impl runnable having run() deciding when to call what method of EvenOddPrint ]

Task#1 -> Write a EvenOddPrint Class 
- flag to decide when to wait else notify and print
-sync printEven() and printOdd() using wait notifyAll


 public class EvenOddPrint {
    
    boolean flag = false;

    synchronized void printEven(int number) {

        if (flag == false) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("Even:" + number);
        flag = false; //once flag is false the odd() will have to print and wait while flag is true
        notifyAll();
    }

    synchronized void printOdd(int number) {
        if (flag == true) {
            try {
                wait();
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        System.out.println("Odd:" + number);
        flag = true;
        notifyAll();
    }
}


TASK #2 - Write a Task Class implementing Runnable having run () having logic to call which of the above print methods

class Task implements Runnable {

    private int max;
    private EvenOddPrint print;
    private boolean isEvenNumber;

    Task(EvenOddPrint print, int max, boolean isEvenNumber) {
        this.print = print;
        this.max = max;
        this.isEvenNumber = isEvenNumber;
    }

    @Override
    public void run() {

        System.out.println("Run method");
        int number = isEvenNumber == true ? 2 : 1;
        while (number <= max) {
            //LOGIC to call which of the print methods
            if (isEvenNumber) {
                // System.out.println("Even : "+
                // Thread.currentThread().getName());
                print.printEven(number);
            } else {
                // System.out.println("Odd : "+
                // Thread.currentThread().getName());
                print.printOdd(number);
            }
            number += 2;
        }

    }

}


TASK#3 - Test class
which creates Thread > passes Task (max,EvenOddPrint, isEvenflag) > call start() which will call tasks run() and begin execution

public class Test {

    public static void main(String ... args){
        EvenOddPrint print = new EvenOddPrint();
        Thread odd = new Thread(new Task(print, 10,  false));
        Thread even = new Thread(new Task(print, 10, true));
        odd.start();
        even.start();
    }

}
 

Click here to View more...

Producer Consumer using Custom BlockingQueue

public class MyBlockingQueue { //Instead use internal BlockingQueue of concurrent package
  
    private Object lock = new Object();
    private List queue = new LinkedList();
    private int limit = 10;

    public MyBlockingQueue(int limit) {
        this.limit = limit;
    }

    public void put(Object item) throws InterruptedException {    // Synchronize with a LOCK -> wait while queueSize == queueLimit
        synchronized (lock) {
            while (this.queue.size() == this.limit) {
                lock.wait();
            }
            if (this.queue.size() == 0) { //if while putting queue is EMPTY -> NotifyAll Producer Threds
                lock.notifyAll();
            }
            this.queue.add(item);
        }
    }

    public Object take() throws InterruptedException {
        synchronized (lock) {
            while (this.queue.size() == 0) {//while removing if size=0 -> lock.wait
                lock.wait();
            }
            if (this.queue.size() == this.limit) {//while removing size = limit -> Notify all Threds
                lock.notifyAll();
            }

            return this.queue.remove(0);
        }
    }
}


public class Producer implements Runnable{
   
   private final MyBlockingQueue sharedQueue;

   public Producer(MyBlockingQueue sharedQueue) {
       this.sharedQueue = sharedQueue;
   }

   @Override
   public void run() {
       for(int i=1; i<11 b="" i="">
           try {
               System.out.println("Produced: " + i);
               sharedQueue.put(i);
           } catch (InterruptedException ex) {
               System.out.println("InterruptedException Occured inside Producer");
           }
       }
   }

}


public class Consumer implements Runnable{
   
    private final MyBlockingQueue sharedQueue;
   
    public Consumer(MyBlockingQueue sharedQueue) {
        this.sharedQueue = sharedQueue;
    }

    @Override
    public void run() {
        while(true){
            try {
                System.out.println("~~~~~~~Consumed: "+ sharedQueue.take());
            } catch (InterruptedException ex) {
                 System.out.println("InterruptedException Occured inside Consumer");
            }
        }
      
    }

}



public class ProducerConsumerTest {

    public static void main(String[] args) {
      
          //Either use internal BlockingQueue or write a custom MyBlockingQueue
         BlockingQueue sharedQueue = new LinkedBlockingQueue();
       
         MyBlockingQueue  myBlockingQueue = new MyBlockingQueue(10);
         

         Thread producerThread = new Thread(new Producer(myBlockingQueue));
       
         Thread consumerThread = new Thread(new Consumer(myBlockingQueue));
                
       
         consumerThread.start();
         producerThread.start();



    }

}

Click here to View more...

How does ConcurrentHashMap work internally

Data Structure

ConcurrentHashMap main motive is to allow concurrent access to the map. HashTable offers synchronized access to the map but the entire map is locked to perform any sort of operation.

In ConcurrentHashMap, the basic strategy is to subdivide the table among segments so that the lock is applied only on a segment rather than the entire table. Each segment manages its own internal hash table in size 2^x >=(capacity/no. of segments).

Locking is applied only for updates. In case of of retrievals, it allows full concurrency, retrievals reflect the results of the most recently completed update operations. 

If we assume four threads are going to concurrently work on a map of initial capacity 32, we would want the table to be partitioned into four segments, each managing a hash table of capacity 8. 
ConcurrentHashMap Data Structure would be:


 



For comparison sake, below figure shows the data structure of a HashMap. 

 

Segment

The collection maintains a list of 16 segments by default, each of which is used to guard (or lock on) a single bucket of the map. This effectively means that 16 threads can modify the collection at a single time (as long as they’re all working on different buckets). This level of concurrency can be increased using the optional concurrencyLevel constructor argument. 

public ConcurrentHashMap(int initialCapacity,
                         float loadFactor, int concurrencyLevel)

The maximum size it can go up to is 2¹⁶. Greater the concurrency level, greater would be the no. of segments and lesser the size of hash table that the segment manages. Using a significantly higher value than you need can waste space and time, and a significantly lower value can lead to thread contention. 

Put Entry

New Entry

ConcurrentHashMap doesn't allow NULL values. The key cannot be NULL, it is used to locate the segment index and then the actual bucket within the segment. The key's hash is re-hashed to strengthen the hash value to defend against poor quality hash functions. This is important as the hash will be used to locate both segment and hash table index. The upper bits will be used to locate the segment index and the lower bits to locate the table index within the segment. This is why re-hashing function is different from that of HashMap as the hash bits need to be spread better to regularize both segment and index locations. 

    private static int hash(int h) {
        h += (h <>> 10);
        h += (h <>>  6);
        h += (h <<   2) + (h <>> 16);
    }

 

Put If Absent

ConcurrentMap offers new method putIfAbsent() which is similar to put except that the value will not be overridden if key already exists. This is equivalent to 

if (!map.containsKey(key))
    return map.put(key, value);
else
   return map.get(key);

except that the action is performed atomically. 

containsKey is not synchronized so the above code may cause unexpected behavior. Lets look into the below scenario: 

1. Thread A is trying to remove a key, calls remove(key), acquires lock.
2. Thread B tries to execute above code, calls containsKey(key). The key exists as Thread A has not yet released the lock so the old value is returned.
3. Thread A removes the key and releases the lock.

The above scenario proves that the code is not atomic as it returned a key which it thinks exists but actually doesn't. 
Also, performance wise it is slow, containsKey has to locate the segment and traverse through the table to find the key. Method put needs to do the same traversing to locate the bucket and put the value. The new methodputIfAbsent is a bit faster as it avoids double traversing. It goes through the same internal flow as put, avoids overriding the old value if key already exists and simply returns the old value. 

Class Diagram

Below class diagram shows that ConurrentHashMap is made up of Segments which in turn is made up ofHashEntries. 


 

Remove Entry

HashEntry is an immutable class so the next link can't be adjusted. All entries following removed node stays in the list, but all preceding ones need to be cloned. 

HashEntry newFirst = e.next;
for (HashEntry p = first; p != e; p = p.next)
newFirst = new HashEntry(p.key, p.hash, newFirst, p.value);
tab[index] = newFirst;

 

Re-sizing

Since power-of-two expansion is used, the elements from each bin must either stay at same index, or move with a power of two offset. 

int idx = e.hash & sizeMask;

For a capacity of 16 elements, sizeMask would be 1111 so the lower 4 bits of hash will form the index. 
When the capacity is doubled, sizeMask would be 11111 and now the lower 5 bits of hash will form the index. 

So the index would be same if the 5^th is 0 else the index would be incremented by 2⁴. 
The trailing consecutive sequence nodes can be re-used instead of unnecessary node creation if they they point to the same slot after the re-hashing. Here in the below example the trailing nodes in blue hues will all end up in the same index. 



 

ConcurrentHashMap example:- 

import java.util.HashMap;

import java.util.Iterator;

import java.util.Map;

import java.util.concurrent.ConcurrentHashMap;

public class ConcurrentHashMapTest {

      public static void main(String[] args) {

            // ConcurrentHashMap implementation details

            Map concurrentHMap = new ConcurrentHashMap();

            concurrentHMap.put("FIRST", "1");

            concurrentHMap.put("SECOND", "2");

            concurrentHMap.put("THIRD", "3");

            concurrentHMap.put("FOURTH", "4");

            concurrentHMap.put("FIFTH", "5");

            concurrentHMap.put("SIXTH", "6");

            System.out.println("ConcurrentHashMap before iteration : " + concurrentHMap);

            Iterator ccHMapIterator = concurrentHMap.keySet().iterator();

            while (ccHMapIterator.hasNext()) {

                  String key = ccHMapIterator.next();

                  if (key.equals("FOURTH"))

                        concurrentHMap.put(key + "-GAURAV", "I AM NEW KEY IN CCHMAP");

            }

            System.out.println("ConcurrentHashMap after iteration : " + concurrentHMap);

            System.out.println("**************************************");

            // HashMap implementation details

            Map hMap = new HashMap();

            hMap.put("FIRST", "1");

            hMap.put("SECOND", "2");

            hMap.put("THIRD", "3");

            hMap.put("FOURTH", "4");

            hMap.put("FIFTH", "5");

            hMap.put("SIXTH", "6");

            System.out.println("HashMap before iteration : " + hMap);

            Iterator hashMapIterator = hMap.keySet().iterator();

            while (hashMapIterator.hasNext()) {

                  String key = hashMapIterator.next();

                  if (key.equals("FOURTH"))

                        hMap.put(key + "-SHIVAM", "I AM NEW KEY IN HMAP");

            }

            System.out.println("HashMap after iteration : " + hMap);

      }

}

Result:-

ConcurrentHashMap before iteration : {THIRD=3, FIRST=1, SIXTH=6, FOURTH=4, FIFTH=5, SECOND=2}

ConcurrentHashMap after iteration : {THIRD=3, FIRST=1, SIXTH=6, FOURTH-GAURAV=I AM NEW KEY IN CCHMAP, FOURTH=4, FIFTH=5, SECOND=2}

**************************************

HashMap before iteration : {FIFTH=5, THIRD=3, FOURTH=4, SECOND=2, SIXTH=6, FIRST=1}

Exception in thread "main" java.util.ConcurrentModificationException

      at java.util.HashMap$HashIterator.nextEntry(HashMap.java:793)

      at java.util.HashMap$KeyIterator.next(HashMap.java:828)

      at com.gaurav.collection.ConcurrentHashMapTest.main(ConcurrentHashMapTest.java:43)

ConcurrentHashMap advantages summary

•  ConcurrentHashMap allows concurrent read and thread-safe update operation. All operations of ConcurrentHashMap are thread-safe.

•  With update operation, ConcurrentHashMap only lock a specific part of Map instead of whole Map.

• Concurrent update is achieved by internally dividing Map into small segments which is defined using concurrency level.

• Iterator returned by ConcurrentHashMap is fail safe and never throw ConcurrentModificationException. 

• ConcurrentHashMap doesn’t allow null as key or value.

Click here to View more...

Creating and Starting Threads - 3 ways

1) Thread Subclass
2) Runnable Interface Implemention
3) Callable (call) and Futures (get) - BEST and Latest (with help of Executors(submit (callable obj))

Method 1

public class MyThread extends Thread {
    
    public void run(){
       System.out.println("MyThread running"); 
    }
  }

MyThread myThread = new MyThread();
myTread.start(); 

OR

You can also create an anonymous subclass of Thread like this:

  Thread thread = new Thread(){
    public void run(){
      System.out.println("Thread Running");  
    }
  }
   thread.start();

Method 2

public class MyRunnable implements Runnable {

    public void run(){
       System.out.println("MyRunnable running");
    }
  }

   Thread thread = new Thread(new MyRunnable());
   thread.start(); 

OR

You can also create an anonymous implementation of Runnable, like this:

   Runnable myRunnable = new Runnable(){

     public void run(){
        System.out.println("Runnable running");
     }
   }
   Thread thread = new Thread(myRunnable);
   thread.start();

Method 3: 

 Using Callable and Futures - refer next post

Subclass or Runnable?

There are no rules about which of the two methods that is the best. Both methods works. Personally though, I prefer implementing Runnable, and handing an instance of the implementation to a Thread instance. When having the Runnable's executed by a thread pool it is easy to queue up the Runnable instances until a thread from the pool is idle. This is a little harder to do with Thread subclasses.
Sometimes you may have to implement Runnable as well as subclass Thread. For instance, if creating a subclass of Thread that can execute more than one Runnable. This is typically the case when implementing a thread pool.

Common Pitfall: Calling run() instead of start()

When creating and starting a thread a common mistake is to call the run() method of the Thread instead of start(), like this:
Thread newThread = new Thread(MyRunnable());
thread.run(); //should be start();
At first you may not notice anything because the Runnable's run() method is executed like you expected. However, it is NOT executed by the new thread you just created. Instead the run() method is executed by the thread that created the thread. In other words, the thread that executed the above two lines of code. To have the run() method of the MyRunnable instance called by the new created thread, newThread, you MUST call the newThread.start() method.

Click here to View more...

Callable and Future

Callable interface represents a thread that returns a value. You can use Callable objects to do calculation.
Callable is a generic interface:

 
interface Callable
  

V indicates the type of data returned by the task.
Callable defines only one method, call( ):

V call( ) throws Exception

You define the task that you want performed in call( ) method.
After that task completes, you return the result. If the result cannot be computed, call( ) must throw an exception.
A Callable task is executed by submit( ) method from ExecutorService.
It is shown here:

 
 Future submit(Callable task)
  

task is the Callable object that will be executed in its own thread.
The result is returned through an object of type Future.
Future is a generic interface that represents the value returned by a Callable object. Future is defined like this:

  interface Future
  

V specifies the type of the result.
To obtain the returned value, you will call Future's get( ) method, which has these two forms:

V get( ) throws InterruptedException, ExecutionException

waits for the result indefinitely.

V get(long wait, TimeUnit tu) throws InterruptedException, ExecutionException, TimeoutException

specify a timeout period to wait.

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class Main {
  public static void main(String args[]) {
    ExecutorService es = Executors.newFixedThreadPool(3);

    Future<Integer> f = es.submit(new Sum());
    try {
      System.out.println(f.get());
    } catch (Exception exc) {
      System.out.println(exc);
    }
    es.shutdown();
  }
}
class Sum implements Callable<Integer> {
  Sum() {
  }

  public Integer call() {
    int sum = 0;
    for (int i = 1; i <= 100; i++) {
      sum += i;
    }
    return sum;
  }
}
  

Click here to View more...

Atomic Variables in JAVA

When a data (typically a variable) can be accessed by several threads, you must synchronize the access to the data to ensure visibility and correctness.

By example, if you have a simple counter (yes, once again) :

1	public class Counter {

2	private int value;

3

4	public int getValue(){

5	return value;

6

}

7

8	public int getNextValue(){

9	return value++;

10

}

11

12	public int getPreviousValue(){

13	return value--;

14

}

15

}

This class works really well in single-threaded environment, but don’t work at all when several threads access the same Counter instance.You can solve the problem using synchronized at method level :

1	public class SynchronizedCounter {

2	private int value;

3

4	public synchronized int getValue(){

5	return value;

6

}

7

8	public synchronized int getNextValue(){

9	return value++;

10

}

11

12	public synchronized int getPreviousValue(){

13	return value--;

14

}

15

}

This class now works well. But locking is not a lightweight mechanism and have several disadvantages. When several threads try to acquire the same lock, one or more threads will be suspended and they will be resumed later. When the critical section is little, the overhead is really heavy especially when the lock is often acquired and there is a lot of contention. Another disadvantage is that the other threads waiting of the lock cannot do something else during waiting and if the thread who has the lock is delayed (due to a page fault or the end of the time quanta by example), the others threads cannot take their turn.

So how to do to avoid this disadvantages ? We must use non-blocking algorithms. This algorithms don’t use blocking mechanisms and by that fact are more scalable and performing. These algorithms use low-level machine instructions which are atomic to ensure the atomicity of higher-level operations. While locking is a pessimistic approach, we can also use optimistic technique to develop algorithms. This time, we’ll detect collisions between threads in which case, the operation fails and we do something else (often retrying the same operation).

The actual processors provide several instructions that simplify greatly the implementation of these non-blocking algorithms, the most-used operation today is the compare-and-swap operation (CAS). This operation takes three parameters, the memory address, the expected current value and the new value. It atomically update the value at the given memory address if the current value is the expected, otherwise it do nothing. In both cases, the operation return the value at the address after the operation execution. So when several threads try to execute the CAS operation, one thread wins and the others do nothing. So the caller can choose to retry or to do something else. We often use this operation to implement another operation, the compare-and-set. This method makes exactly the same things as CAS but return a boolean indicating if the operation succeeded or not.

Before Java 5.0, this operation was not available directly to developer, but in Java 5.0 several atomic variables (for int, long, boolean and reference values) were added. The int and long versions also supports numeric operations. The JVM compiles these classes with the better operations provided by the hardware machine, CAS or a Java implementation of the operation using a lock. Here are the classes :

• AtomicInteger
• AtomicLong
• AtomicBoolean
• AtomicReference

All these classes supports compare-and-set (via the compareAndSet() method) and other operations (get(), set() and getAndSet()). The setters operations are implemented using compareAndSet. These classes supports multi-threaded access and have a better scalability than synchronizing all the operations.

Here is how we can rewrite our counter using an AtomicInteger :

1	public class AtomicCounter {

2	private final AtomicInteger value = new AtomicInteger(0);

3

4	public int getValue(){

5	return value.get();

6

}

7

8	public int getNextValue(){

9	return value.incrementAndGet();

10

}

11

12	public int getPreviousValue(){

13	return value.decrementAndGet();

14

}

15

}

The incrementAndGet() and decrementAndGet() methods are two of the numeric operations provided by the AtomicLong and AtomicInteger classes. You also have getAndDecrement(), getAndIncrement(), getAndAdd(int i) and addAndGet().

This version is faster than the synchronized one and is also thread safe.

If you only have the compareAndSet(), here is how we can implement increment() method using it :

1	public void increment(AtomicInteger integer){

2	while(true){

3	int current = integer.get();

4	int next = current + 1;

5

6	if(integer.compareAndSet(current, next)){

7

return;

8

}

9

}

10

}

This seems to be complicated, but this is the cost of non-blocking algorithms. When we detect collision, we retry until the operation succeeded. This is the common schema for non-blocking algorithms.

Here is a thread-safe Stack implemented using AtomicReference :

1	public class Stack {

2	private final AtomicReference head = new AtomicReference(null);

3

4	public void push(String value){

5	Element newElement = new Element(value);

6

7	while(true){

8	Element oldHead = head.get();

9	newElement.next = oldHead;

10

11	//Trying to set the new element as the head

12	if(head.compareAndSet(oldHead, newElement)){

13

return;

14

}

15

}

16

}

17

18	public String pop(){

19	while(true){

20	Element oldHead = head.get();

21

22	//The stack is empty

23	if(oldHead == null){

24	return null;

25

}

26

27	Element newHead = oldHead.next;

28

29	//Trying to set the new element as the head

30	if(head.compareAndSet(oldHead, newHead)){

31	return oldHead.value;

32

}

33

}

34

}

35

36	private static final class Element {

37	private final String value;

38	private Element next;

39

40	private Element(String value) {

41	this.value = value;

42

}

43

}

44

}

It’s really more complicated than using synchronized on the two methods but also more performing if there is contention (and often even if there is no contention).

So this ends this post. To conclude, atomic variables classes are a really good way to implement non-blocking algorithms and moreover are also a very good alternative to volatile variables, because they can provide atomicity and visibility.

Click here to View more...