Java Multi threading
Benefit: simple , responsive, performance then processes.Costs - complex design , context switching overhead, resources consumption. ( suppose 100 threads in system seating idle .. sleeping/ waiting)
Shared state - variable , memory , database...
problems - deadlock , race condition, critical section
Concurrency Models and Distributed Systems
Concurrency models uses threads and distributed systems uses processes.
Synchronization
- to avoid race condition.
- to avoid race condition.
- Each object has a lock.
- lock objects.
- Independent objects can have different lock.
Synchronization Methods:
volatile keyword.
Synchronization method and Synchronized block.
wait and notify.
better Solution:
variables stored in threads local stack..
Local Object References - All objects are stored in the shared heap.
If localObject reference passed to some method, it will create locally instance and will be thread safe. .
volatile keyword.
Synchronization method and Synchronized block.
wait and notify.
Race Condition and Critical Section:
Different threads trying to access data so output will be wrong.
Critical Section:
public class Counter {
protected long count = 0;
public void add(long value){
this.count = this.count + value;
}
}
Critical Section Throughput:
public class TwoSums {
private int sum1 = 0;
private int sum2 = 0;
public void add(int val1, int val2){
synchronized(this){
this.sum1 += val1;
this.sum2 += val2;
}
}
}
better Solution:
public class TwoSums {
private int sum1 = 0;
private int sum2 = 0;
private Integer sum1Lock = new Integer(1);
private Integer sum2Lock = new Integer(2);
public void add(int val1, int val2){
synchronized(this.sum1Lock){
this.sum1 += val1;
}
synchronized(this.sum2Lock){
this.sum2 += val2;
}
}
}
Thread Safety and Shared Resources-
Local variables - primitive variables thread safe.variables stored in threads local stack..
public void someMethod(){
long threadSafeInt = 0;
threadSafeInt++;
}
Local Object References - All objects are stored in the shared heap.
If localObject reference passed to some method, it will create locally instance and will be thread safe. .
public void someMethod(){
LocalObject localObject = new LocalObject();
localObject.callMethod();
method2(localObject);
}
public void method2(LocalObject localObject){
localObject.setValue("value");
}
Object Member Variables:
public class NotThreadSafe{
StringBuilder builder = new StringBuilder();
public add(String text){
this.builder.append(text);
}
}
below code will Not be thread Safe because two thread running on same object.
NotThreadSafe sharedInstance = new NotThreadSafe(); new Thread(new MyRunnable(sharedInstance)).start(); new Thread(new MyRunnable(sharedInstance)).start();
however if threads running on different objects, code will be thread safe.
new Thread(new MyRunnable(new NotThreadSafe())).start(); new Thread(new MyRunnable(new NotThreadSafe())).start();
Thread Safety and Immutability:
public class ImmutableValue{
private int value = 0;
public ImmutableValue(int value){
this.value = value;
}
public int getValue(){
return this.value;
}
}
Note: No function to setValue().Return new instance if to preserve immutability.
public class ImmutableValue{
private int value = 0;
public ImmutableValue(int value){
this.value = value;
}
public int getValue(){
return this.value;
}
public ImmutableValue add(int valueToAdd){
return new ImmutableValue(this.value + valueToAdd);
}
}
The Reference is not Thread Safe!
public void add(int newValue){
this.currentValue = this.currentValue.add(newValue);
}
Java Memory Models:
Java internal java memory model-
Java internal java memory model-
1. A local variable may be of a primitive type, in which case it is totally kept on the thread stack.
2. A local variable may also be a reference to an object. In that case the reference (the local variable) is stored on the thread stack, but the object itself if stored on the heap.
3. An object may contain methods and these methods may contain local variables. These local variables are also stored on the thread stack, even if the object the method belongs to is stored on the heap.
4. An object's member variables are stored on the heap along with the object itself. That is true both when the member variable is of a primitive type, and if it is a reference to an object.
5. Static class variables are also stored on the heap along with the class definition.
6. Objects on the heap can be accessed by all threads that have a reference to the object. When a thread has access to an object, it can also get access to that object's member variables. If two threads call a method on the same object at the same time, they will both have access to the object's member variables, but each thread will have its own copy of the local variables.
public class MyRunnable implements Runnable() {
public void run() {
methodOne();
}
public void methodOne() {
int localVariable1 = 45;
MySharedObject localVariable2 =
MySharedObject.sharedInstance;
//... do more with local variables.
methodTwo();
}
public void methodTwo() {
Integer localVariable1 = new Integer(99);
//... do more with local variable.
}
}
public class MySharedObject {
//static variable pointing to instance of MySharedObject
public static final MySharedObject sharedInstance =
new MySharedObject();
//member variables pointing to two objects on the heap
public Integer object2 = new Integer(22);
public Integer object4 = new Integer(44);
public long member1 = 12345;
public long member1 = 67890;
}
Note : http://tutorials.jenkov.com/java-concurrency/java-memory-model.html
Thread Signaling:
1. Signaling via shared Objects:
public class MySignal{
protected boolean hasDataToProcess = false;
public synchronized boolean hasDataToProcess(){
return this.hasDataToProcess;
}
public synchronized void setHasDataToProcess(boolean hasData){
this.hasDataToProcess = hasData;
}
}
- setting the signal values in some shared object variable.
- set the boolean member variable hasDataToProcess to true from inside a synchronized block
2. Busy Wait
Thread B which is to process the data is waiting for data to become available for processing.
protected MySignal sharedSignal = ...
...
while(!sharedSignal.hasDataToProcess()){
//do nothing... busy waiting
}
- No CPU utilization.
3. wait(), notify() and notifyAll()
public class MonitorObject{
}
public class MyWaitNotify{
MonitorObject myMonitorObject = new MonitorObject();
public void doWait(){
synchronized(myMonitorObject){
try{
myMonitorObject.wait();
} catch(InterruptedException e){...}
}
}
public void doNotify(){
synchronized(myMonitorObject){
myMonitorObject.notify();
}
}
}
- This is mandatory to call from synchronized block. A thread cannot call wait(), notify() or notifyAll() without holding the lock on the object the method is called on. If it does, an IllegalMonitorStateException is thrown.
Missing Signals:
- notify() and notifyAll() do not save the method calls to them in case no threads are waiting when they are called. The notify signal is then just lost.
- if a thread calls notify() before the thread to signal has called wait(), the signal will be missed by the waiting thread
- in some cases this may result in the waiting thread waiting forever, never waking up, because the signal to wake up was missed.
public class MyWaitNotify2{
MonitorObject myMonitorObject = new MonitorObject();
boolean wasSignalled = false;
public void doWait(){
synchronized(myMonitorObject){
while(!wasSignalled){ //do not use if cond'n here
try{
myMonitorObject.wait();
} catch(InterruptedException e){...}
}
//clear signal and continue running.
wasSignalled = false;
}
}
public void doNotify(){
synchronized(myMonitorObject){
wasSignalled = true;
myMonitorObject.notify();
}
}
}
Note : Don't call wait() on constant String's or global objects
ie : String myMonitorObject = "";
- The problem with calling wait() and notify() on the empty string, or any other constant string is, that the JVM/Compiler internally translates constant strings into the same object.
- That means, that even if you have two different MyWaitNotify instances, they both reference the same empty string instance.
- This also means that threads calling doWait() on the first MyWaitNotify instance risk being awakened by doNotify() calls on the second MyWaitNotify instance.
DEADLOCK
Prevention:
Locks in java:
A lock is a thread synchronization mechanism like synchronized blocks except locks can be more sophisticated than Java's synchronized blocks
1. Simple Locks:
synchronized block:public class Counter{
private int count = 0;
public int inc(){
synchronized(this){
return ++count;
}
}
}
Now using locks instead synchronized block:
public class Lock{
private boolean isLocked = false;
public synchronized void lock()
throws InterruptedException{
while(isLocked){
wait();
}
isLocked = true;
}
public synchronized void unlock(){
isLocked = false;
notify();
}
}
here is the lock class implementation:
public class Lock{
private boolean isLocked = false;
public synchronized void lock()
throws InterruptedException{
while(isLocked){
wait();
}
isLocked = true;
}
public synchronized void unlock(){
isLocked = false;
notify();
}
}
Calling unlock() From a finally-clause
lock.lock();
try{
//do critical section code, which may throw exception
} finally {
lock.unlock();
}
This little construct makes sure that the
Lock is unlocked in case an exception is thrown from the code in the critical section.
If
unlock() was not called from inside a finally-clause, and an exception was thrown from the critical section, the Lock would remain locked forever, causing all threads calling lock() on that Lock instance to halt indefinately.Semaphores:
A Semaphore is a thread synchronization construct that can be used either to send signals between threads to avoid missed signals, or to guard a critical section like you would with a lock. Java 5 comes with semaphore implementations in the
java.util.concurrent package.
Simple implemenatation:
public class Semaphore {
private boolean signal = false;
public synchronized void take() {
this.signal = true;
this.notify();
}
public synchronized void release() throws InterruptedException{
while(!this.signal) wait();
this.signal = false;
}
}
Using Semaphores for signalling:
Semaphore semaphore = new Semaphore(); SendingThread sender = new SendingThread(semaphore); ReceivingThread receiver = new ReceivingThread(semaphore); receiver.start(); sender.start();
public class SendingThread {
Semaphore semaphore = null;
public SendingThread(Semaphore semaphore){
this.semaphore = semaphore;
}
public void run(){
while(true){
//do something, then signal
this.semaphore.take();
}
}
}
public class RecevingThread {
Semaphore semaphore = null;
public ReceivingThread(Semaphore semaphore){
this.semaphore = semaphore;
}
public void run(){
while(true){
this.semaphore.release();
//receive signal, then do something...
}
}
}
The
take() method sends a signal which is stored internally in the Semaphore. The release() method waits for a signal.
When received the signal flag is cleared again, and the
release() method exited.
Counting Semaphores:
public class CountingSemaphore {
private int signals = 0;
public synchronized void take() {
this.signals++;
this.notify();
}
public synchronized void release() throws InterruptedException{
while(this.signals == 0) wait();
this.signals--;
}
}
Bounded Semaphore
The
CoutingSemaphore has no upper bound on how many signals it can store. We can change the semaphore implementation to have an upper boundpublic class BoundedSemaphore {
private int signals = 0;
private int bound = 0;
public BoundedSemaphore(int upperBound){
this.bound = upperBound;
}
public synchronized void take() throws InterruptedException{
while(this.signals == bound) wait();
this.signals++;
this.notify();
}
public synchronized void release() throws InterruptedException{
while(this.signals == 0) wait();
this.signals--;
this.notify();
}
}
Using Semaphores as Locks
It is possible to use a bounded semaphore as a lock. To do so, make upper bound to 1.
Blocking Queue Implementation
The implementation of a blocking queue looks similar to a Bounded Semaphore. Here is a simple implementation of a blocking queue:
public class BlockingQueue {
private List queue = new LinkedList();
private int limit = 10;
public BlockingQueue(int limit){
this.limit = limit;
}
public synchronized void enqueue(Object item)
throws InterruptedException {
while(this.queue.size() == this.limit) {
wait();
}
if(this.queue.size() == 0) {
notifyAll();
}
this.queue.add(item);
}
public synchronized Object dequeue()
throws InterruptedException{
while(this.queue.size() == 0){
wait();
}
if(this.queue.size() == this.limit){
notifyAll();
}
return this.queue.remove(0);
}
}
Notice how
notifyAll() is only called from enqueue() and dequeue() if the queue size is equal to the size bounds (0 or limit). If the queue size is not equal to either bound when enqueue() or dequeue() is called, there can be no threads waiting to either enqueue or dequeue items.Thread Pool:
Instead of starting a new thread for every task to execute concurrently, the task can be passed to a thread pool. As soon as the pool has any idle threads the task is assigned to one of them and executed. Internally the tasks are inserted into a Blocking Queue which the threads in the pool are dequeuing from. When a new task is inserted into the queue one of the idle threads will dequeue it successfully and execute it. The rest of the idle threads in the pool will be blocked waiting to dequeue tasks.
public class ThreadPool {
private BlockingQueue taskQueue = null;
private List<PoolThread> threads = new ArrayList<PoolThread>();
private boolean isStopped = false;
public ThreadPool(int noOfThreads, int maxNoOfTasks){
taskQueue = new BlockingQueue(maxNoOfTasks);
for(int i=0; i<noOfThreads; i++){
threads.add(new PoolThread(taskQueue));
}
for(PoolThread thread : threads){
thread.start();
}
}
public synchronized void execute(Runnable task) throws Exception{
if(this.isStopped) throw
new IllegalStateException("ThreadPool is stopped");
this.taskQueue.enqueue(task);
}
public synchronized void stop(){
this.isStopped = true;
for(PoolThread thread : threads){
thread.doStop();
}
}
}
public class PoolThread extends Thread {
private BlockingQueue taskQueue = null;
private boolean isStopped = false;
public PoolThread(BlockingQueue queue){
taskQueue = queue;
}
public void run(){
while(!isStopped()){
try{
Runnable runnable = (Runnable) taskQueue.dequeue();
runnable.run();
} catch(Exception e){
//log or otherwise report exception,
//but keep pool thread alive.
}
}
}
public synchronized void doStop(){
isStopped = true;
this.interrupt(); //break pool thread out of dequeue() call.
}
public synchronized boolean isStopped(){
return isStopped;
}
}
Lalit Kumar Solanki
No comments:
Post a Comment