Sincronizzazione dei Thread in Java

Daniela Tibaldi: dtibaldi@deis.unibo.it

1) Sottoclasse della classe Thread

Thread possiede un metodo run() che la sottoclasse deve ridefinire• si crea un’istanza della sottoclassetramite new• si esegue un thread chiamando il metodostart() che a sua volta richiama il metodo run()

Esempio di classe SimpleThread che èsottoclasse di Thread:

public class SimpleThread extends Thread{

public SimpleThread(String str){super(str);}public void run(){ for (int i = 0; i < 10; i++){ System.out.println(i +" "+ getName());

try{sleep((int)(Math.random()*1000));}catch (InterruptedException e){}

}System.out.println("DONE! " + getName());

}}public class TwoThreadsTest{

public static void main (String[] args){ new SimpleThread("Jamaica").start();new SimpleThread("Fiji").start();

}}

E se occorre definire thread che non siano necessariamente

sottoclassi di Thread?

Thread come classe che implementa l’interfaccia runnable

• implementare il metodo run() nella classe che implementa l’interfaccia Runnable

• creare un’istanza della classe tramite new• creare un’istanza della classe Thread con

un’altra new, passando come parametro l’istanza della classe che si è creata

• invocare il metodo start() sul thread creato, producendo la chiamata al suo metodo run()

Interfaccia Runnable: maggiore flessibilità thread come sottoclasse di qualsiasi altra classe

Esempio di classe EsempioRunnable che implementa l’interfaccia Runnable ed è sottoclasse

di MiaClasseclass EsempioRunnable extends MiaClasse implements Runnable{

// non e’ sottoclasse di Threadpublic void run() {

for (int i=1; i<=10; i++)System.out.println(i + “ ” + i*i);

}}public class Esempio{

public static void main(String args[]){

EsempioRunnable e = new EsempioRunnable();Thread t = new Thread(e);t.start();

}}

Metodi per il controllo di thread• start() fa partire l’esecuzione di un thread. La macchina

virtuale Java invoca il metodo run() del thread appena creato

• stop() forza la terminazione dell’esecuzione di un thread. Tutte le risorse utilizzate dal thread vengono immediatamente liberate (lock inclusi), come effetto della propagazione dell’eccezione ThreadDeath

• suspend() blocca l'esecuzione di un thread in attesa di una successiva operazione di resume(). Non libera le risorse impegnate dal thread (lock inclusi)

• resume() riprende l'esecuzione di un thread precedentemente sospeso. Se il thread riattivato ha una priorità maggiore di quello correntemente in esecuzione, avrà subito accesso alla CPU, altrimenti andrà in coda d'attesa

• sleep(long t) blocca per un tempo specificato (time) l'esecuzione di un thread. Nessun lock in possesso del thread viene rilasciato.

• join() blocca il thread chiamante in attesa della terminazione del thread di cui si invoca il metodo. Anche con timeout

• yield() sospende l'esecuzione del thread invocante, lasciando il controllo della CPU agli altri thread in coda d'attesa

I metodi precedenti interagiscono con il gestore della sicurezza della macchina virtuale Java (SecurityManager, checkPermission())

Il problema di stop()e suspend()stop() e suspend() rappresentano azioni “brutali” sul ciclo di vita di un thread rischio di determinare situazioni

inconsistenti o di blocco critico (deadlock)

• se il thread sospeso aveva acquisito una risorsa in maniera esclusiva, tale risorsa rimane bloccata e non è utilizzabile da altri, perché il thread sospeso non ha avuto modo di rilasciare il lock su di essa

• se il thread interrotto stava compiendo un insieme di operazioni su risorse comuni, da eseguirsi idealmente in maniera atomica, l’interruzione può condurre ad uno stato inconsistente del sistema

JDK 1.4, pur supportandoli ancora per ragioni di back-compatibility, sconsiglia l’utilizzo dei metodi stop(), suspend() e resume() (metodi deprecated)

Si consiglia invece di realizzare tutte le azioni di controllo e sincronizzazione fra thread tramite i metodi wait(), notify() e notifyAll() (astrazione di monitor)

Sincronizzazione di Threads

Quando due o più thread eseguono concorrentemente, è (in generale) impossibile prevedere l'ordine in cui le loro istruzioni saranno eseguite.

Nel caso in cui i thread invochino metodi sullo stesso oggetto SONO POSSIBILI INCONSISTENZE!

Mutua EsclusioneDifferenti thread che fanno parte della stessa applicazione Java condividono lo stesso spazio di memoria

è possibile che più thread accedano contemporaneamente allo stesso metodo o alla stessa sezione di codice di un oggetto

Servono meccanismi di mutua esclusione sull’accesso al metodo e alla sezione di codice condivisa

JVM supporta la mutua esclusione nell’accesso a risorse condivise tramite la keyword synchronized

synchronized su:• singolo metodo• blocco di istruzioni

In pratica:

• a ogni oggetto Java è automaticamente associato un lock

• per accedere a un metodo o una sezione synchronized, un thread deve prima acquisire il lock dell’oggetto

• il lock è automaticamente rilasciato quando il thread esce dalla sezione synchronized, o se viene interrotto da un’eccezione

• un thread che non riesce ad acquisire un lock rimane sospeso sulla richiesta della risorsa fino a che il lock non è disponibile

NOTA:ad ogni oggetto contenente metodi o blocchi synchronized viene assegnata una sola coda di attesa ( principale differenza con i monitor: mancanza di variabili condizione)

Due thread non possono accedere contemporaneamente a due sezioni synchronizeddiverse di uno stesso oggetto

Quando un thread in esecuzione tenta di accedere ad una sezione synchronized di un oggetto il cui lock è già stato acquisito da un altro thread, che esegue concorrentemente, esso si mette in attesa di poter acquisire il lock.

Il thread rimane in stato runnable entry-set

Appena il lock è rilasciato, lo scheduler può potenzialmente mettere in esecuzione il thread.

Sincronizzazione

Mutua esclusione uso di lock (tramite keywordsynchronized)

Politiche di sincronizzazione nell’accesso ad una risorsa condivisa: uso di costrutti synchronized e primitive wait(), notify() e notifyAll()

Ogni oggetto Java (istanza di una sottoclasse qualsiasi della classe java.lang.Object) fornisce i metodi di sincronizzazione wait(), notify() e notifyAll()

• wait()blocca l’esecuzione del thread invocante in attesa che un altro

thread invochi i metodi notify() o notifyAll() per quell’oggetto. Il lock viene rilasciato.

Il thread invocante deve essere in possesso del lock sull’oggetto (ossia la wait() deve essere eseguita all’interno di un metodo o sezione synchronized) altrimenti viene lanciata un’eccezione.

Il blocco del thread invocante avviene in maniera atomica dopo aver rilasciato il lock. Il thread riacquisisce il lock prima di ritornare dal metodo wait() per evitare situazioni di corsa critica. Anche varianti con specifica di timeout.

Quando un thread esegue una wait() si porta in stato Not-Runnablenon può essere messo in esecuzione dallo scheduler. wait-set

• notify()risveglia un unico thread in attesa sul monitor dell’oggetto in

questione. Ciò significa che il thread risvegliato viene portato in stato Runnable e può quindi essere messo in esecuzione dallo scheduler.

Se più thread sono in attesa, la scelta avviene in maniera arbitraria, dipendente dall’implementazione della macchina virtuale Java. Il thread risvegliato compete con ogni altro thread, come di norma, per ottenere la risorsa protetta. Anche notify() deve essere chiamata all’interno di un metodo synchronized.

• notifyAll()si comporta come notify(), ma risveglia tutti i thread in

attesa per l’oggetto in questione. È necessario tutte le volte in cui più thread possono essere sospesi su differenti sezioni critiche dello stesso oggetto oppure per diverse cause (unica coda d’attesa).

Per ogni oggetto sincronizzato esiste un’unica coda di attesa (wait-set).

Esempio (Controllore per un parcheggio di auto)

Progettare un sistema di controllo per l’ingresso ad un parcheggio per auto. Il sistema permette alle auto di entrare solo quando il parcheggio non è pieno e non permette alle auto di uscire quando non ci sono auto nel parcheggio.Simulare l’arrivo e l’uscita di auto tramite threads.

Threads: Arrivals, Departures (implementano Runnable).CarParkControl è la classe che fornisce il controllo nell’accesso al parcheggio.

class CarParkControl {protected int spaces;protected int capacity;CarParkControl(int capacity) {capacity = spaces = n;}synchronized void arrive() {… --spaces; …

}synchronized void depart() {… ++spaces; …

}}class Arrivals implements Runnable {CarParkControl carpark;Arrivals(CarParkControl c) {carpark = c;}public void run() {try {while(true) {carpark.arrive();

}} catch (InterruptedException e){}

}}

e le condizioni di sospensione?class CarParkControl {protected int spaces;protected int capacity;CarParkControl(int capacity){capacity = spaces = n;}

synchronized void arrive() throws InterruptedException {while (spaces==0) wait();--spaces;notifyAll();

}synchronized void depart() throws InterruptedException {while (spaces==capacity) wait();++spaces;notifyAll();

}}

Ricapitolando…

Il parcheggio è l’oggetto sincronizzato ed esiste un’unica coda di attesa sia per i thread rappresentativi delle auto in ingresso sia per quelle in uscita al parcheggio.

condizioni di sospensione diverse.

• E se per testare la condizione di sospensione avessimo utilizzato if(condizione)wait()?

• E se avessi utilizzato la notify() invece che la notifyAll()?

• c1, c2, c3 e c4 eseguono il metodo depart() quando il parcheggio è vuoto

• p1 running esegue il metodo arrive()

• p2, p3 e p4 cercano di eseguire il metodo arrive si sospendono in attesa dell’acquisizione del lock

• p1 termina: esegue la notify() e rilascia il lock

parcheggio

parcheggio entry-set wait-set

p2 p3p4

c1c2 c3c4

entry-set wait-set

p2 p3p4c1

c2 c3c4

• p2 acquisisce il lock. Esegue il metodo arrive()

• p2 termina: esegue la notify() e rilascia il lock

parcheggio entry-set wait-set

p3p4c1c2

c3c4

Il parcheggio è pieno.

parcheggio entry-set wait-set

p3p4c1c2

c3c4

• p3 acquisisce il lock. Esegue il metodo arrive() parcheggio pieno wait()

parcheggio entry-set wait-set

p4c1c2

p3c3c4

parcheggio entry-set wait-set

c1c2

p4p3c3c4

• p4 acquisisce il lock. Esegue il metodo arrive() parcheggio pieno wait()

• c1 acquisisce il lock. Esegue il metodo depart()

• c1 termina: esegue la notify() e rilascia il lock

parcheggio entry-set wait-set

c2c3

p4p3c4

• c2 acquisisce il lock. Esegue il metodo depart()

• c2 termina: esegue la notify() e rilascia il lock

parcheggio entry-set wait-set

c3c4

p4p3

parcheggio entry-set wait-set

c2c3

p4p3c4

Deadlock!!!

• c3 acquisisce il lock. Esegue il metodo depart() parcheggio vuoto wait()

• c4 acquisisce il lock. Esegue il metodo depart() parcheggio vuoto wait()

public class Mailbox { private int[]contenuto;private int contatore, testa, coda;

public mailbox(){contenuto = new int[N];contatore = 0;testa = 0;coda = 0;

}public synchronized int preleva (){

int elemento;while (contatore == 0) wait();elemento = contenuto[testa];testa = (testa + 1)%N;--contatore; notifyAll(); return elemento;

} public synchronized void deposita (int valore){

while (contatore == N) wait(); contenuto[coda] = valore;coda = (coda + 1)%N;++contatore; notifyAll();

} }

Esempio: Buffer circolare

Altro Esempio

public class GuardedBoundedBuffer implements Buffer {private List data;private final int capacity;public GuardedBoundedBuffer(int capacity) {

data = new ArrayList(capacity);this.capacity = capacity;

}public synchronized Object take() throws Failure {

while (data.size() == 0)try { wait(); }catch(InterruptedException ex) {throw new Failure();}Object temp = data.get(0);data.remove(0);notifyAll();return temp;

}public synchronized void put(Object obj) throws Failure {

while (data.size() == capacity)try { wait(); }catch(InterruptedException ex) {throw new Failure();}data.add(obj);notifyAll();

}public synchronized int size() {

return data.size(); }public int capacity() { return capacity; }}

Principale limitazione dei meccanismi Java:

unica wait-set per un oggetto sincronizzato

difficoltà di sospendere thread su differenti code, mancanza dei semafori privati

Produttori e Consumatori Concorrenti

Possibili Politiche di Accesso

• se ci sono dei lettori già attivi sulla risorsa, può un altro thread lettore accedere anche se c’è già uno scrittore sospeso?

• se sulla risorsa è attivo uno scrittore e in attesa ci sono lettori e scrittori, a chi dare la precedenza? Ai lettori? Agli scrittori? Al primo che ha fatto la richiesta? Random? In maniera alternata?

Esempio:class ReadersWriters extends Rwbasic {private int nr=0;private synchronized void startRead() { nr++; }public synchronized void endRead() {nr--;if (nr==0) notify(); // sveglia i Writers in attesa

}public void read() {// il metodo read non è più synchronizedstartRead();System.out.println(“wrote “ +data);endRead();

}public synchronized void write() {while (nr>0) //finché ci sono dei readers

try { wait();}catch (InterruptedException ex) {return;}

data++;System.out.println(“wrote “ +data);notify; //necessaria per gli altri writers in attesa

}}

class Reader extends Thread {int rounds; ReadersWriters rw;public Reader(int rounds, ReadersWriters rw){ this.rounds=rounds; this.rw=rw;}public void run() {for (int i=0; i< rounds; i++) {rw.read();}

}}class Writer extends Thread {int rounds; ReadersWriters rw;public Writer(int rounds, ReadersWriters rw){ this.rounds=rounds; this.rw=rw;}public void run() {for (int i=0; i<rounds; i++) {rw.write();}

}}class Main {static ReadersWriters rw; rw=new ReadersWriters ();public static void main(String[] args) {int rounds=Integer.parseInt(args[0], 10);new Reader(rounds, rw).start();new Writer(rounds, rw).start();

}} A chi viene data la priorità nell’accesso alla risorsa condivisa (variabile data)?

Variante:public abstract class RW {protected int activeReaders = 0;protected int activeWriters = 0;protected int waitingReaders = 0;protected int waitingWriters = 0;public void read() {beforeRead();try { doRead(); } finally { afterRead(); }

}public void write() {beforeWrite();try {doWrite(); } finally { afterWrite();}

}protected boolean allowReader() {return waitingWriters == 0 && activeWriters == 0;

}protected boolean allowWriter() {return activeReaders == 0 && activeWriters == 0;

}

protected synchronized void beforeRead() {++waitingReaders;while (!allowReader()) try { wait(); }catch (InterruptedException ex){ ... }

--waitingReaders; ++activeReaders;}

protected abstract void doRead();protected synchronized void afterRead() {--activeReaders; notifyAll();

}protected synchronized void beforeWrite() {++waitingWriters;while (!allowWriter())try { wait(); }catch (InterruptedException ex){ ... }

--waitingWriters; ++activeWriters;}protected abstract void doWrite();protected synchronized void afterWrite() {--activeWriters; notifyAll();

}}

Semafori in JavaUtilizzando i costrutti primitivi di Java si possono realizzare meccanismi piùsofisticati (semafori e anche monitor).

public class Semaphore {private int value;public Semaphore (int initial){

value = initial;}synchronized public void V()

++value;notify();

}synchronized public void P()

throws InterruptedException{while (value == 0) wait();--value;

}}

Produttore/Consumatore con Semafori Javaclass SemaBuffer implements Buffer {

protected Object[] buf; protected int in = 0;protected int out = 0; protected int count = 0;protected int size = 0;Semaphore full; //numero dei messaggiSemaphore empty; //numero degli spaziSemaBuffer(int size) {this.size=size; buf=new Object[size];full = new Semaphore(0); empty = new Semaphore(size);

}}synchronized public void put(Object o)

throws InterruptedException {empty.P(); buf[in] = o; ++count; in=(in+1)%size;full.V();

}synchronized public Object get() throws InterruptedException{full.P(); Object o =buf[out]; buf[out]=null;--count; out=(out+1)%size; empty.V(); return (o);

}Funziona tutto bene???

L’esempio sopra non garantisce liveness, perché ci può essere deadlock:

• se il buffer è vuoto• il consumatore fa una get• il consumatore si blocca e rilascia il lock sul semaforo• il produttore invoca put ma viene sospeso sul lock del buffer, che non è stato rilasciato dal consumatore.

In questo caso il deadlock è causato da chiamate innestate (nestedmonitor).

Esempio corretto

public void put(Object o) throws InterruptedException {empty.P();synchronized(this){buf[in] = o; ++count; in=(in+1)%size;

}full.V();

}Il deadlock può essere evitato solo attraverso un progetto attento, sincronizzando la regione critica di accesso al buffer solo dopo il passaggio del semaforo.

Quindi la realizzazione di costrutti di sincronizzazione di alto livello in Javarichiede particolare attenzione.