Concurrent Programming - Threads

Java language natively supports threads, which are several sequences of instructions running in parallel, but belonging to the same application. An example would be a server which accepts and manages several connections at the same time. There are two ways of implementing a thread:

• by extending class java.lang.Thread;
• by implementing interface java.lang.Runnable.

Class Thread as well as interface Runnable define a method called run(). This method is the start point of the new thread, just like public static void main(String[]) for the main thread. This method run() can be called in two ways:

• directly calling it, in which case it is ran on the same thread, as normal method;
• calling method start(), which starts a new thread and starts executing method run() in parallel.

public class PrintThread extends Thread{

    private int index;

public PrintThread(int _index){
    index = _index;
}

public void run(){
    for(int i=0; i<5; i++){
        System.out.println("This is thread " + index);
        try{
            //pause for 0.5 seconds (500 ms)
            Thread.sleep(500);
        }catch(InterruptedException _ie){
            System.out.println(_ie.getMessage());
        }
    }
}

}

This is the normal call of method run(). The program will first display the test on thread 1 five times, then the one on thread 2 five times, then the one on thread 3, etc:

public class NormalStarter{

public static void main(String[] _args){
    for(int i=0; i<5; i++){
        PrintThread _thread = new PrintThread(i + 1);
        _thread.run();
    }
}

}

This is the behavior when calling start(). The program will display texts from all threads at the same time:

public class ThreadStarter{

public static void main(String[] _args){
    for(int i=0; i<5; i++){
        PrintThread _thread = new PrintThread(i + 1);
        _thread.start();
    }
}

}

Same thing by using the Runnable interface:

public class PrintRunnable implements Runnable{

    private int index;

public PrintRunnable(int _index){
    index = _index;
}

public void run(){
    for(int i=0; i<10; i++){
        System.out.println("This is thread " + index);
        try{
            //pause for 1 second (1000 ms)
            Thread.sleep(1000);
        }catch(InterruptedException _ie){
            System.out.println(_ie.getMessage());
        }
    }
}

}

public class ThreadStarterRunnable{

public static void main(String[] _args){
    for(int i=0; i<5; i++){
        // polymorphism: PrintRunnable is a Runnable
        Runnable _runnable = new PrintRunnable(i + 1);

        PrintThread _thread = new Thread(_runnable);
        _thread.start();
    }
}

}

Synchronizing Threads - Semaphores

There are situations in multithreading applications, where multiple threads are accessing a common resource. In some of these situations, if some methods are program areas using these common resources are accessed by multiple threads at the same time, certain events occur that result in the malfunction of the application. These areas need to be executed atomically, meaning that during their execution by a thread, no other thread can interrupt the current thread or execute the same area.

In the virtual machine, this is achieved with a system of monitors. A monitor is a hidden field defined in class Object (so that by inheriting it, any object can be used as a monitor) which keeps track of the threads that access one or more code sections. These sections are called synchronized. If a thread is about to enter a synchronized section, then the virtual machine verifies if the monitor is not locked, meaning that no other thread executes instructions in any section synchronized by that monitor. The the monitor is locked (if another thread has it locked), then the current thread will stop and wait for its release. If the monitor is unlocked, the thread will lock it and it will continue execution.

In Java, the keyword for synchronizing a code section is synchronized.

public class Stack{

    public static final int MAX_STACK_SIZE = 128;
    
    private Object[] stack;
    private int stackTop;

public Stack(){
    this(MAX_STACK_SIZE);
}

public Stack(int _maxSize){
    stack = new Object[_maxSize];
    stackTop = 0;
}

public synchronized boolean empty(){
    return stackTop == 0;
}

public synchronized boolean full(){
    return stackTop == stack.length;
}


public Object pop() throws Exception{
    synchronized(this){
        if(!empty()){
            return stack[--stackTop];
        }
    }

    throw new Exception("Stack empty");
}

public void push(Object _obj) throws Exception{
    synchronized(this){
        if(!full()){
            stack[stackTop++] = _obj;
        }
    }

    throw new Exception("Stack full");
}

}

In the example above, if two threads call methods push(Object) and pop() at the same time and they would not be synchronized, situations might occur when the counter stackTop would be incremented by one thread, then decremented by the second before the first finishes writing in the array stack, effectively writing in the wrong position. By using keyword synchronized, we are safe to assume that none of the methods declared as such and none of the synchronized sections will be executed at the same time by different threads.

Rule:A thread can not execute a synchronized section belonging to another thread BUT it can execute a synchronized section that already belongs to it. As an example, the synchronized section in method push(Object) is calling method full(), which is itself synchronized by the same monitor.

If a method is declared synchronized, this is equivalent of declaring the whole method content in a synchronized(this) block. Therefore, a method declared synchronized is automatically synchronized using the current object as a monitor.

If a static method is declared synchronized, this is equivalent as declaring the whole method content in a synchronized(<class_name>.class) block. Therefore, a static method declared synchronized is automatically synchronized using the object of type Class associated with the current class as a monitor.

public class StaticSync{

public static synchronized void printSmth(){
    System.out.println("Smth");
}

public static void printSmthElse(){
    synchronized(StaticSync.class){
        System.out.println("SmthElse");
    }
}

}

Race Conditions - Barriers

In the example above, is there are two threads, one writing elements onto the stack and one consuming them, we can see that if any of the threads is slightly faster, it will end up throwing an exception, either because the stack gets empty or full. This is what's called a race condition: one more code sections, executed by two or more threads that are expected to be executed in a particular order that ends up being inverted. So in the example, if the stack is empty, then the thread pushing should take the monitor before the one popping, or an exception will be thrown.

This problem is solved by using a barrier system, which is a system that stops a thread in a certain spot, until a condition is met. In Java, this is done by calling methods wait() / wait(long) / wait(long, int) and notify() / notifyAll() belonging to class Object.

Methods wait() / wait(long) / wait(long, int) block the current thread until another thread calls one of the notify() / notifyAll() methods for the same object, until the time given as an argument has expired or until another thread interrupts the waiting thread by calling Thread.interrupt().

Rule:Calls to wait() methods are always done in a synchronized section and are called for the monitor object.

public class Stack{

    public static final int MAX_STACK_SIZE = 128;
    
    private Object[] stack;
    private int stackTop;

public Stack(){
    this(MAX_STACK_SIZE);
}

public Stack(int _maxSize){
    stack = new Object[_maxSize];
    stackTop = 0;
}

public synchronized boolean empty(){
    return stackTop == 0;
}

public synchronized boolean full(){
    return stackTop == stack.length;
}


public Object pop(){
    synchronized(this){
        while(empty()){
            try{
                this.wait();
            }catch(InterruptedException _ie){
                  System.out.println("InterruptedException: " + _ie.getMessage());
            }
        }
        this.notifyAll();
        return stack[--stackTop];
    }
}

public void push(Object _obj){
    synchronized(this){
        while(full()){
            try{
                this.wait();
            }catch(InterruptedException _ie){
                  System.out.println("InterruptedException: " + _ie.getMessage());
            }
        }
        stack[stackTop++] = _obj;
        this.notifyAll();
    }
}

}

Methods of type notify() reactivate waiting threads. These don't have to be called from within a synchronized section, but it is recommended to do so. They are called for the monitor objects used to call wait on the waiting threads.

A thread reactivated using notify needs to wait for the monitor unlocking in order to enter the synchronized section, like any other thread. Notice this behavior in method pop(), where notifyAll() has been called, before extracting the element from the stack, on the previous line. The program functions correctly because no other thread will start execution until the current thread exists the synchronized area.

External resources:

1. http://docs.oracle.com/javase/tutorial/essential/concurrency/
2. http://www.tutorialspoint.com/java/java_multithreading.htm