multi threadingjava

Threads: the OriginCallableFuture

ThreadPoolSynchronisation 1Synchronisation

ForkJoin

Multithreading en JavaEléments conceptuels de programmation concurrente

Elvadas NONO

Softeam Cadextan , Startech Java

October 16, 2013

1 / 20



ForkJoin

Outline1 Threads: the Origin2 Callable3 Future4 ThreadPool

ExecutorExecutorServiceExecutors

5 Synchronisation 1SynchronizedReentrantLockReadWriteLock

6 SynchronisationCountDownLatchCyclicBarrierPhaser

7 ForkJoin

2 / 20



ForkJoin

Thread

Thread: Plus petite unité d'exécution dans un programme.

Application: Ensemble de Threads s'exécutant en parallèledans la JVM

Non Deamon Thread vs Deamon Thread, Main Thread

Notion de priorité d'un Thread

Deux façons de créer un Thread:

extends java.lang.Thread ClassImplement java.lang.Runnable Interface

3 / 20



ForkJoin

/∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗ OPTION1∗∗∗∗∗∗∗∗∗∗∗∗∗/public class MonThread extends Thread {

public void run() {// Thread behavior goes here !! . . .

}}

// Start ThreadMonTread t1= new MonThread();t1. start () ;

/∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗∗ OPTION2∗∗∗∗∗∗∗∗∗∗∗∗∗/public class MyRunnable implements Runnable {

public void run() {// Thread behavior goes here !! . . .

}}

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

4 / 20



ForkJoin

Problèmes

Et le résultat ? retour de la methode run: void

gestion des exceptions

démarrage du thread

synchrone ou asynchrone?

4 / 20



ForkJoin

L'interface Callable

public Interface Callable<V>

java.util.concurrent package, since JDK 1.5

Dé�nit une unique methode sans argument call.

Retourne un résultat Générique VPeut lever une exception

5 / 20



ForkJoin

public class MyCallable implements Callable<String> {

public String call() throws exception {

// Thread behavior goes here!! . . .

}

}

6 / 20



ForkJoin

Future

L'interface Future: Résultat d'une opération asynchrone qui setermine dans le futur

Méthodes fournies

boolean cancel(boolean mayInterruptIfRunning)V get() Attends la �n du traitement et récupère le resultatboolean isCancelled() indique si la tache a été annuléeboolean isDone() Retourne Vrai si la le traitement est terminé

6 / 20



ForkJoin


Executor

L'interface Executor: Découpler la soumission des tâchesconcurrentes des mécanismes d'exécution de chaque tâche

Méthodes fournies

void execute(Runnable command) Exécuter la commandepassée en paramètre dans le futureFuture<v> submit = executor.submit(worker);List<Future<T>�> invokeAll(Collection<? extendsCallable<T>�> tasks) Lien Future/Callable/Executor

Variantes

ExecutorServiceThreadPoolExeceutorSerial, Parallel, Cache, etcc

7 / 20



ForkJoin


ExecutorService

ExecutorService: Spécialisation de l'interface Executor a�n desoumettre des

RunnableCallableCollection de CallableCollection de RunnableSous ensemble de Runnable/Callable

8 / 20



ForkJoin


Executors

Factory de ThreadPool / Executor

static ExecutorService newSingleThreadExecutor() Créé unexécuteur o�ciant avec un seul worker thread et une queue detraitement.static ExecutorService newFixedThreadPool(int nThreads)pool de nThread Threads faisant usage d'une �le de tâches.static ExecutorService newCachedThreadPool() Poolfonctionnant suivant le sytème de cachestatic ScheduledExecutorServicenewSingleThreadScheduledExecutor() Executeur pouvantlancer des threads périodiques, avec des délais

9 / 20

Threads for the OriginCallableFuture


ForkJoin


public class PoolThreadTest {

private static final int NB_TASKS = 100;

public static void main(String args[]){

// Execultor service Factory

ExecutorService cacheExecutor =

Executors.newCachedThreadPool();

ExecutorService singleExecutor =

Executors.newSingleThreadExecutor();

ExecutorService fixedExecutor =

Executors.newFixedThreadPool(3);

// Create a task list

Collection<DummyCallable> tasks= new

9 / 20

Threads executor the OriginCallableFuture


ForkJoin


ArrayList<>(NB_TASKS);

for(int i=0;i<NB_TASKS;i++){ DummyCallable

t= new DummyCallable(i);

tasks.add(t);

}

//start executor

startExecutor(fixedExecutor, tasks);

}

private static void startExecutor(ExecutorService

fixedExecutor, Collection<DummyCallable> tasks) {

try {

List<Future<Long>>

9 / 20



ForkJoin


futures=executor.invokeAll(tasks);

executor.awaitTermination(1, TimeUnit.MINUTES); }

catch (InterruptedException |

IllegalStateException e) {

e.printStackTrace();

}finally{

executor.shutdown();

System.out.println("bye bye!");

}

}

10 / 20



ForkJoin


ScheduledExecutorService

Exécuter des tâches

Après un délai

schedule (Callable task, long delay, TimeUnit timeunit)schedule (Runnable task, long delay, TimeUnit timeunit)

A une fréquence �xe: example toutes les 5min

scheduleAtFixedRate (Runnable, long initialDelay, long period,TimeUnit timeunit)

Avec un délai �xe entre la �n d'un Run et le début du suivant

scheduleWithFixedDelay (Runnable, long initialDelay, longperiod, TimeUnit timeunit)

Comme tous les Executeurs vous devez les arrêterexplicitement après usage

shutdown() or shutdownNow()10 / 20



ForkJoin

SynchronizedReentrantLockReadWriteLock

ScheduledExecutorService ses =

Executors.newScheduledThreadPool(5);

ScheduledFuture sf = ses.schedule(new Callable() {

public Object call() throws Exception {

System.out.println("Executed!");

return "StarTech Java!";

} }, 5, TimeUnit.SECONDS);

System.out.println("result = " + sf.get());

ses.shutdown();

11 / 20



ForkJoin


Synchronized

Protection de l'accès aux sections critiques

Implémentation d'un lock structuré:synchronized methods andsynchronized statements

Interdit l'exécution concurrente de regions protégées sur unmême objet par deux Threads

Reentrant mechanism

11 / 20



ForkJoin


//***********SOLUTION1

public static synchronized Singleton getInstance(){

if (instance==null) //1

instance=new Singleton(); //2

return instance; //3

}

//*************SOLUTION2

public static Singleton getInstance(){

if (instance==null) { //1

synchronized (Singleton.class){ //2

instance=new Singleton();

}

}

return instance;

}

11 / 20



ForkJoin


// Q:which one is correct? A: None Why?

12 / 20



ForkJoin


Synchronized issues

Problèmes

Thread automatiquemnt bloqué si le verrou est occupé

Pas de vue sur la �le des Threads bloqués et reprise après lock

Gestion des locks multiples

Lock Structuré

Unfair Lock

No way to check ability to trying for lock interruptibly

No delay/timeout waiting period

Di�erence Read/Write operations

12 / 20



ForkJoin


ReetrantLock

ReetrantLock: Lock Reentrant plus �exible

Fairness: The most waiting Thread is always choosen ifactivatedSpecialisation: ReaderLock and WritterLockWaiting Thread QueuetryLock to check if lock is available : not blockant

13 / 20



ForkJoin


ReadWriteLock

ReadWriteLock: Interface de gestion d'un lock enLecture/Ecriture

Plusieurs Lecteurs autorisés en parallèle

Les Rédacteurs sont prioritaires sur les lecteurs

En interne: Deux lock séparés

ReentrantReadWriteLock implements ReadWriteLock

14 / 20



ForkJoin

CountDownLatchCyclicBarrierPhaser

ReadWriteLock readWriteLock = new

ReentrantReadWriteLock();

readWriteLock.readLock().lock();

// Plusieurs Lecteur peuvent entrer dans cette section

// si le verrou n'est pas occupé par un Redacteur, et

aucun rédacteur en Attente du verrou

readWriteLock.readLock().unlock();

readWriteLock.writeLock().lock();

// seuls les Redacteurs peuvent atteindre cette

section,

// si aucun lecteur actif.

readWriteLock.writeLock().unlock();

15 / 20



ForkJoin


CountDownLatch

Permettre à un ou plusieurs Thread d'attendre la �n del'exécution d'un ensemble d'actions par un ou plusieurs autresThreads

un CountDownLatch est initialisé avec un compteur

la méthode countDown() décrémente le compteur

la méthode await() bloque le Thread appelant jusqu'a ce quele compteur atteigne la valeur zero

Tous les Thread bloqués sont libérés lorsque le compteurdevient nul

le compteur ne peut pas être réinitialisé.

Tout appel à await après que le compteur soit passé à zero setermine imédiatement

15 / 20

Threads the OriginCallableFuture


ForkJoin


final CountDownLatch countDownLatch = new

CountDownLatch(2);

Thread t = new Thread(new Runnable() {

public void run() {

System.out.println(countDownLatch.getCount());

//(2)

countDownLatch.countDown();


//(1)

try {

Thread.sleep(5000);

countDownLatch.countDown();


//(0)

} catch (InterruptedException e) {


15 / 20



ForkJoin


}

}

});

t.start();

try {

countDownLatch.await(); //blocks the current Thread

System.out.println("MainThread

"+countDownLatch.getCount()); // run only when latch=0

} catch (InterruptedException e) {


}

16 / 20



ForkJoin


CyclicBarrier

Permettre à un ou plusieurs Thread de s'attendre à un pointde rendez vous.

un CyclicBarrier est initialisé avec un compteur

Contrairement au CountDownLatch, le compteur peut êtreréinitialisé pour un nouveau cycle: reset

la méthode await() bloque le Thread appelant jusqu'a ce quel'une des conditions suivantes soit véri�ée

le compteur atteind la valeur zero naturellement ou par unappel à resetLe Thread appelant est interrompu par un autre Threadun Thread en attente du RDV atteind son timeout

Tous les Thread bloqués sont libérés lorsque le compteurdevient nul ou le compteur ressetté

Possibilité de dé�nir un traitement à executer lorsque tous lesThreads sont au point de RendezVous

16 / 20



ForkJoin


//create a Cyclic Barrier

CyclicBarrier barrier = new CyclicBarrier(5);

//Make the current thread wait at barrier

barrier.await();

//after 15s the Thread is release even if the counter is

not 0

barrier.await(15, TimeUnit.SECONDS);

//define a barrier action

Runnable barrierAction = new Runnable( ... ;

CyclicBarrier barrier = new CyclicBarrier(3,

barrierAction);

17 / 20



ForkJoin


Phaser

Mécanisme de synchornisation à un point de rendez voussimilaire au CyclicBarrier

Supporte plusieurs cycles, JDK 7

le nombre de parties n'est pas �xé à l'avance: dynamique au �ldu temps et des cycles

l'enregistrement se fait à tout moment: methode register

Atteindre le point de RDV int arrive() / et se désister duprochain cycle arriveAndDeregister()

boolean onAdvance(int phase, int registeredParties), réagir auxchangement de cycles

int arriveAndAwaitAdvance() attendre les autres Threads aupoint de rendez vous.

Attendre une phase particulière int awaitAdvance(int phase)17 / 20



ForkJoin

Fork/Join

Programmation Multi-Core, DPR algorithm,JSR 166 libs,native in JDK7

ForkJoinPool

Fixed Pool Size Thread, if no size speci�ed=> Number ofavailable processorsExecutorService permettant d'exécuter des ForkJoinTasksUn Thread peut récupérer une tâche dans la �le d'attente d'unautre Thread

fork: subdiviser un traitement en plusieurs sous traitementsparallèles

RecursiveTask & RecursiveAction: sous classes deForkJoinTask

join: Récupérer le résultat d'une sous tâche 18 / 20

Threads public the OriginCallableFuture


ForkJoin

18 / 20



ForkJoin

public class MaxSearchTask extends RecursiveTask{

protected Long compute() {

if(sup-inf<5){

return simpleMax(); }

else{

int mid=(inf+sup)/2;

MaxSearchTask search1 = new MaxSearchTask(inf,mid, t);

MaxSearchTask search2 = new MaxSearchTask(mid, sup, t);

invokeAll(search1,search2);

return Math.max((Long)search1.join(),(Long)

search2.join());

}

...

}

19 / 20



ForkJoin

Multithreading is also

Types Atomiques, AtomicInteger, AtomicXXX

Volatile: The volatile force the thread to update the originalvariable.

Utiliser les collections appropriées en fonction des situations:

Thread safe collections vs non Thread safe collectionsBlockingQueue

Outils de pro�ling: JPro�ler, VisualVM,

Debugging, JDWP

...

19 / 20



ForkJoin

Q&A

Now this is not the end.

It is not even the beginning of the end.

But it is, perhaps, the end of the beginning.

Winston Churchill

20 / 20

multi threadingjava

Documents

threadpool executor

thread class

runnable callable collection

thread behavior

invokealltasks executor

future futurev

thread t2

dlaischedule callable