new - Dipartimento di Matematica e Informatica

Sincronizzazione dei Thread
Vitaliano Milanese
Dipartimento di Matematica e Informatica
Università di Udine
Programmazione Object Oriented
::
Lezione 22
2
THREAD
Sincronizzazione
Ogni insieme di thread cooperanti nell’ esecuzione di un’ applicazione
possono richiedere di essere sincronizzati
in corrispondenza del verificarsi di due situazioni
• accesso ad una stessa risorsa
• interscambio dati necessari alla prosecuzione dei loro task
Seppure similari
• le due situazioni vanno trattate in modo differenziato
• possono richiedere modelli di sincronizzazione diversificati
3
THREAD
Condivisione di Risorse
Una risorsa condivisa costituisce un’ entità
dotata di meccanismi di protezione all’ accesso da parte dei thread
In Java
il meccanismo di protezione
è fornito da un dispositivo di mutua esclusione (mutex)
Per operare su risorse condivise un thread deve:
• acquisire il mutex della risorsa, bloccandone l’ accesso agli altri thread
• accedere alla risorsa mediante i metodi della stessa
• rilasciare il mutex, liberalizzandone l’ accesso agli altri thread
4
THREAD
Condivisione di Risorse
• ogni oggetto è depositario di un dispositivo di mutex
• in condizioni usuali il mutex di un oggetto è ignorato
Tuttavia
se più thread intendono accedere ai metodi di uno stesso oggetto risorsa
devono gestirne il relativo mutex
• il controllo sul mutex è operato da metodi synchronized
• tutti i metodi della classe devono essere synchronized
Il modificatore synchronized è applicabile a
• metodi dinamici (di uno stesso oggetto)
l’ attivazione di uno di essi blocca le richieste di esecuzione dei rimanenti metodi
quando inoltrate alla stessa istanza
• metodi statici (di una stessa classe)
le esecuzioni sono serializzate in modo che nessuno di essi sia interrotto
dall’ esecuzione dei rimanenti
5
THREAD
Condivisione di Risorse
La risorsa vettoreStati di classe VettoreStatiThread costituisce risorsa condivisa
#1, m
Thd1
#4, k
.. t + dt
vettoreStati
t ..
Thd4
Metodi sincronizzati (accesso via mutex)
Stati generati:
(#1 , m) (#4 , k)
Metodi non sincronizzati (mutex ignorato)
Possibili stati generati:
(#4 , m) (? , k)
6
THREAD
Condivisione di Risorse
DA RICORDARE
• Quandi si utilizza il mutex di un oggetto
i campi dello stesso devono essere dichiarati private
• Se un thread accede ad un metodo sincronizzato di un oggetto
è possibile che quel metodo attivi altri metodi sincronizzati dello stesso oggetto
In questo caso ogni chiamata ad altro metodo
causa un incremento del contatore di lock del mutex
7
THREAD
Sincroniz. per Scambio Dati
La sincronizzazione dei thread è necessaria per serializzare
le operazioni di trasferimento dati fra thread mittente/i e destinatario/i
Il linguaggio
deve disporre di primitive in grado di agire sull’ esecuzione dei thread
forzandone la sospensione e la successiva riattivazione
In Java
• I metodi destinati alla sincronizzazione dei thread sono insiti negli oggetti
• Appartengono alla classe Object da cui derivano le classi di applicazione
8
THREAD
Sincroniz. per Scambio Dati
Dato
(Risorsa)
comunica
Dato
Risorsa
condivisa
sospensione
Sveglia !!!
Thd sospeso
Thd attivo
9
THREAD
Sincroniz. per Scambio Dati
Un thread può portarsi nello stato bloccato (sospeso) per:
• chiamata al metodo sleep()
ritorna runnable a sleeping concluso
oppure per ricezione di un comando interrupt()
Thread interrompibile
• attesa del completamento di un’ operazione di I/O
ritorna runnable a completamento dell’ operazione di I/O
• accesso ad un metodo o ad un blocco synchronized con mutex non disponibile
ritorna runnable a seguito di rilascio del mutex
• chiamata del metodo wait()
Thread interrompibile
ritorna runnable per intercettazione del metodo notify() oppure notifyAll()
10
THREAD
Sincroniz. per Scambio Dati
METODI DI SINCRONIZZAZIONE (Classe Object)
• void wait (long timeout) throws InterruptedException
• void wait() throws InterruptedException
Sospendono il thread correntemente in esecuzione (ev. per il tempo timeout)
Richiedono l’ uso del costrutto try-catch
Un thread sospeso perde il possesso della risorsa su cui opera
e viene inserito nella coda dei processi bloccati e in attesa
pertanto
Le chiamate a wait() vanno incluse in metodi [o regioni] synchronized
I thread sospesi e in attesa del rilascio di una risorsa possono essere risvegliati
da altro thread che opera sulla stessa risorsa con i metodi notify() o notifyAll()
Un thread risvegliato è trasferito nella coda dei processi eseguibili e
compete per ottenere nuovamente il possesso della risorsa condivisa di interesse
11
THREAD
Sincroniz. per Scambio Dati
METODI DI SINCRONIZZAZIONE (Classe Object)
• void notify()
Risveglia un thread che è sospeso in attesa del rilascio di una risorsa
Il thread riottiene il possesso della risorsa
eseguendo un metodo sincronizzato appartenente alla risorsa
• void notifyAll()
Risveglia tutti i thread sospesi in attesa del rilascio di una risorsa
I metodi risvegliati competono per riottenere il controllo della risorsa
in base alla loro priorità e/o al momento della loro sospensione
Il metodo notify() presuppone che tutti i thread sospesi
siano in attesa del verificarsi della stessa condizione
[i.e. siano stati bloccati sullo stesso metodo]
In caso contrario è necessario un controllo esplicito per validare lo stato del thread
12
THREAD
Regioni Critiche
Con il termine
• regione critica
si intende una regione di codice sottoposta a vincoli di sincronizzazione
synchronized (objSinc) {
comandi
}
I comandi sono eseguiti previa acquisizione del mutex di objSinc
da usare quale blocco di sincronizzazione
USO
Rendere sincronizzate operazioni definite da metodi che non sono synchronized
Tipicamente sfruttata per utilizzare in ambito multi-thread classi non protette
rendendole controllate e protette da oggetti di altra classe
13
THREAD
Sincronizzazione
La comunicazione di dati fra thread può seguire modelli diversificati
tra cui:
• produttore/i – consumatore/i
richiede un supporto dati condiviso
più o meno complesso in relazione al numero di thread coinvolti
• monitor
richiede una classe di sincronizzazione
• pipe line
richiede un canale di comunicazione dedicato
14
THREAD
Produttore Consumatore
Schema PRODUTTORE - CONSUMATORE
Uno o pù thread produttori trasferiscono dati
a uno o più thread consumatori (utilizzatori) di quei dati
sfruttando
una risorsa condivisa dotata di metodi sincronizzati
Th1
GUI
Th2
Th3
Th4
Th5
Array di supporto
(condiviso)
Visualizza
le configurazioni
a mano a mano
che sono disponibili
Thread sospeso
e periodicamente
risvegliato
15
THREAD
Produttore Consumatore
Modellazione dei thread produttori di configurazioni:
StatoThread (idThread, countDown)
public class ThreadBanale extends Thread {
private static int threadCount = 0;
private int countDown = 5;
private VettoreStatiThread vettoreStati;
private ThreadGUI threadGUI;
private StatoThread statoThread;
private String idThread;
// Risorsa condivisa
// Thread consumatore
public ThreadBanale (VettoreStatiThread vettoreStati, ThreadGUI threadGUI) {
super ("#" + ++threadCount);
idThread = "#" + threadCount;
this.vettoreStati = vettoreStati;
this.threadGUI = threadGUI;
}
Continua …
16
THREAD
Produttore Consumatore
public class ThreadBanale extends Thread {
……
public void run() {
while (true) {
if (countDown == 0) return;
// Aggiorna l’array comunicando la nuova configurazione di stato
statoThread = new StatoThread (idThread, countDown);
vettoreStati.addStatoThread (statoThread);
// Risveglia il thread di supporto alla GUI
synchronized (threadGUI) {
threadGUI.notify();
}
try { sleep (100); }
catch (InterruptedException e) {}
countDown--;
}
}
}
17
THREAD
Produttore Consumatore
class StatoThread {
private String idThread;
private int countDown;
public StatoThread (String idThread, int countDown) {
this.idThread = idThread;
this.countDown = countDown;
}
public String getIdThread() { return idThread; }
public int getCountDown() { return countDown; }
}
18
THREAD
Produttore Consumatore
class VettoreStatiThread {
Vector<StatoThread> vettoreStati;
public VettoreStatiThread() {
vettoreStati = new Vector<StatoThread> ();
}
public synchronized void addStatoThread (StatoThread statoThread) {
vettoreStati.addElement (statoThread);
}
public synchronized Enumeration getStatiThread() {
Enumeration statiThread = vettoreStati.elements();
vettoreStati = new Vector<StatoThread>();
return statiThread;
}
public synchronized int size() { return vettoreStati.size(); }
}
19
THREAD
Produttore Consumatore
Modellazione del thread consumatore di configurazioni
public class ThreadGUI extends JFrame implements Runnable {
private CtrlThread panThread;
……
private VettoreStatiThread vettoreStati;
// Risorsa condivisa
public ThreadGUI (VettoreStatiThread vettoreStati) {
this.vettoreStati = vettoreStati;
setDefaultCloseOperation (JFrame.EXIT_ON_CLOSE);
setTitle (getClass().getName());
setSize (hSize, vSize);
Container cnt = getContentPane();
panThread = new CtrlThread();
cnt.add (panThread, "Center");
setVisible (true);
}
Continua …
20
THREAD
Produttore Consumatore
public void run() {
while (true) {
synchronized (this) {
try {
wait();
// Sospensione del thread
}
catch (InterruptedException e) { throw new RuntimeException (e); }
}
Risveglio
del
thread
Enumeration enumStati = vettoreStati.getStatiThread();
while (enumStati.hasMoreElements()) {
StatoThread statoThread = (StatoThread) enumStati.nextElement();
String idThread = statoThread.getIdThread();
int countDown = statoThread.getCountDown();
panThread.drawBox (idThread, countDown);
}
}
}
}
Continua …
21
THREAD
Produttore Consumatore
Modellazione della classe di tracciatura
class CtrlThread extends JPanel {
private int nrBox, countDown, xRef, yRef, delta;
private String idThread;
private Color color;
public CtrlThread() { // Inizializzazione di nrBox, idThread, xRef, yRef, delta; }
public void paintComponent (Graphics g) {
if (nrBox >= 0) {
Graphics2D g2 = (Graphics2D) g;
g2.setColor (color);
g2.fill (new Rectangle2D.Double (xRef, yRef, 30, 30));
g2.setColor (Color.black);
Integer valore = new Integer (countDown);
String str = valore.toString();
g2.drawString (str, xRef + 10, yRef + 20);
}
}
Continua …
22
THREAD
Produttore Consumatore
public void drawBox (String idThread, int countDown) {
this.idThread = idThread;
this.countDown = countDown;
nrBox++;
if ((nrBox % 20 ) == 0) {
xRef = 10;
yRef += delta;
}
else
xRef += delta;
if (idThread.equals ("#1"))
color = Color.green;
……
if (idThread.equals ("#5"))
color = Color.yellow;
// Il metodo repaint() va sostituito da
paintImmediately (new Rectangle (xRef, yRef, 30, 30));
}
}
23
THREAD
Produttore Consumatore
public class Pgm_xThread {
public static void main (String[] args) {
// Costruzione della risorsa condivisa
VettoreStatiThread vettoreStati = new VettoreStatiThread();
// Creazione del thread consumatore
ThreadGUI winRunnable = new ThreadGUI (vettoreStati);
Thread winThread = new Thread (winRunnable);
winThread.start();
// Creazione dei thread produttori
for (int k = 0; k < 5; k++) {
ThreadBanale thread = new ThreadBanale (vettoreStati, winRunnable);
thread.start();
}
}
}
24
THREAD
Produttore Consumatore
DA OSSERVARE
Qualora la classe VettoreStatiThread venisse sostituita da quella sotto indicata
class StatoAccessibile {
StatoThread statoThread;
public synchronized void addStatoThread (StatoThread statoThread) {
this.statoThread = statoThread;
}
public synchronized StatoThread getStatoThread() { return statoThread; }
}
il thread consumatore acquisirebbe soltanto parte degli stati generati dai produttori
25
THREAD
Monitor
Schema MONITOR
Uno o più thread produttori trasferiscono dati a uno o più thread consumatori
sfruttando
una risorsa condivisa (monitor) dotata di metodi sincronizzati
avente la funzione di acquisire e smistare i singoli dati trasferiti
A differenza dello schema precedente
• il monitor non fornisce un mero supporto di allocazione
ma
• svolge funzione attiva
acquisendo dai produttori singoli blocchi di dati solo quando disponibile
e smistando tali dati quando posseduti
26
THREAD
Monitor
PROBLEMA
Due thread produttori di dischi (rispettivamente blu e rossi) li devono trasferire ad
un thread consumatore con cadenze temporali prestabilite
TP1
TP2
Monitor
Il monitor
• se vuoto
TCons
acquisisce un dato dal primo thread mittente pronto a trasmettere
cambia il proprio stato
notifica agli altri thread la variazione di stato, eventualmente risvegliandoli
• se pieno
smista il dato posseduto al thread destinatario
cambia il proprio stato
notifica agli altri thread la variazione di stato, eventualmente risvegliandoli
27
THREAD
Monitor
28
THREAD
Monitor
public class Monitor {
private boolean pieno = false;
private Cerchio cerchio;
public synchronized void ricezione (Cerchio c) {
if (pieno)
try {
wait(); // Il produttore che ha acquisito il monitor viene sospeso
}
catch (InterruptedException e) { }
pieno = true;
cerchio = c;
notifyAll(); // Comunica la variazione di stato
}
Continua …
29
THREAD
Monitor
Segue
public synchronized Cerchio rilascio() {
if (!pieno)
try {
wait(); // Il consumatore che ha acquisito il monitor viene sospeso
}
catch (InterruptedException e) { }
pieno = false;
notifyAll(); // Comunica la variazione di stato
return cerchio;
}
}
30
THREAD
public class Produttore implements Runnable {
private Monitor monitor;
private int rgn;
private Frame_xThread frame;
……
public Produttore (Monitor m, int rgn, Frame_xThread frame) {
monitor = m;
this.rgn = rgn;
this.frame = frame;
switch (rgn) {
case 1: coRef = Color.blue;
nMaxCerchi = 8;
break;
case 2: coRef = Color.red;
nMaxCerchi = 5;
break;
}
vtProd = new Vector<Cerchio>();
for (int k = 1; k <= nMaxCerchi; k++) {
cerchio = new Cerchio (0, 0, 10, coRef);
vtProd.addElement (cerchio);
}
Monitor
31
THREAD
Segue
public void run() {
for (int k = 0; k < nMaxCerchi; k++) {
try {
if (rgn == 1)
Thread.sleep (700);
if (rgn == 2)
Thread.sleep (2000);
}
catch (InterruptedException e) { }
cerchio = vtProd.elementAt (0);
monitor.ricezione (cerchio);
vtProd.removeElementAt (0);
frame.modificaRegione (rgn, nMaxCerchi, vtProd);
}
}
public Vector<Cerchio> getElements() { return vtProd; }
}
Monitor
32
THREAD
public class Consumatore implements Runnable {
private Monitor monitor;
private Frame_xThread frame;
private Vector<Cerchio> vtCons;
public Consumatore (Monitor m, Frame_xThread frame) {
monitor = m;
this.frame = frame;
vtCons = new Vector<Cerchio>();
}
public void run() {
while (true) {
Cerchio cerchio = monitor.rilascio();
vtCons.addElement (cerchio);
int nCerchi = vtCons.size();
frame.modificaRegione (0, nCerchi, vtCons);
}
}
}
Monitor
33
THREAD
Monitor
public class Frame_xThread extends JFrame {
private final int RGN_CONS = 0, RGN_PROD1 = 1, RGN_PROD2 = 2;
private RgnFrame rgnCons, rgnProd1, rgnProd2;
public Frame_xThread () {
super ("Produttori - Consumatore");
……
cnt.setLayout (new GridLayout (3, 1, 5, 5));
rgnProd1 = new RgnFrame();
rgnProd1.setBackground (Color.yellow);
cnt.add (rgnProd1);
rgnProd2 = new RgnFrame();
rgnProd2.setBackground (Color.yellow);
add (rgnProd2);
rgnCons = new RgnFrame();
rgnCons.setBackground (Color.lightGray);
add (rgnCons);
setVisible (true);
}
Continua …
34
THREAD
Monitor
public void modificaRegione (int rgn, int nMaxCerchi, Vector<Cerchio> vt) {
switch (rgn) {
case 0: rgnCons.setElements (vt, nMaxCerchi);
rgnCons.repaint();
break;
case 1: rgnProd1.setElements (vt, nMaxCerchi);
rgnProd1.repaint();
break;
case 2: rgnProd2.setElements (vt, nMaxCerchi);
rgnProd2.repaint();
break;
}
}
}
Continua …
35
THREAD
Monitor
class RgnFrame extends JPanel {
private int nMaxCerchi;
private Vector<Cerchio> vt;
public RgnFrame() {
super();
vt = new Vector<Cerchio>();
nMaxCerchi = 0;
}
public void setElements (Vector<Cerchio> vt, int nMaxCerchi) {
this.vt = vt;
this.nMaxCerchi = nMaxCerchi;
}
Continua …
36
THREAD
Monitor
public void paintComponent (Graphics g) {
super.paintComponent (g);
Graphics2D g2 = (Graphics2D) g;
final int cBaseX = -10;
final int cBaseY = 30;
int cX, cY, rg;
Color colore;
final int rgBase = 10;
if (nMaxCerchi != 0) {
cX = cBaseX;
cY = cBaseY;
for (int k = 0; k < vt.size(); k++) {
Cerchio cRef = vt.elementAt (k);
rg = cRef.raggio;
g2.setColor (cRef.colore);
cX = cX + 4*rg;
g2.fill (new Ellipse2D.Double (cX - rg, cY - rg, 2*rg, 2*rg));
}
g2.setColor (Color.lightGray);
rg = rgBase;
for (int k = vt.size() + 1; k < nMaxCerchi+1; k++) {
cX = cX + 4*rg;
g2.draw (new Ellipse2D.Double (cX - rg, cY - rg, 2*rg, 2*rg));
}
}
37
THREAD
Monitor
public static void main (String[] args) {
Frame_xThread winThread = new Frame_xThread();
Monitor monitor = new Monitor();
// Creazione del primo produttore
Produttore prod1 = new Produttore (monitor, 1, winThread);
Thread thProd1 = new Thread (prod1);
Vector<Cerchio> vtProd1 = new Vector<Cerchio>();
vtProd1 = prod1.getElements();
int nCerchiRgnProd1 = vtProd1.size();
winThread.modificaRegione (1, nCerchiRgnProd1, vtProd1);
// Creazione del secondo produttore
Produttore prod2 = new Produttore (monitor, 2, winThread);
Thread thProd2 = new Thread (prod2);
Vector<Cerchio> vtProd2 = new Vector<Cerchio>();
vtProd2 = prod2.getElements();
int nCerchiRgnProd2 = vtProd2.size();
winThread.modificaRegione (2, nCerchiRgnProd2, vtProd2);
Continua …
38
THREAD
// Creazione del consumatore
Consumatore cons = new Consumatore (monitor, winThread);
Thread thCons = new Thread (cons);
// Attivazione dei thread
thProd1.start();
thProd2.start();
thCons.start();
}
Monitor
39
THREAD
Pipeline
Schema PIPELINE
Variante dello schema Produttore Consumatore
Utilizza le classi PipedReader e PipedWriter che realizzano l’ I/O fra thread
mediante pipe (canale di comunicazione)
TP
Usa PipedWriter
per scrivere su pipe
Pipe
TCons
Usa PipedReader
per leggere da pipe
PipedReader va agganciato
a PipedWriter
per realizzare la pipe
Il thread consumatore (contenente PipedReader) è bloccato
quando la pipeline è sotto controllo del produttore (contenente PipedWriter)
oppure è vuota
40
THREAD
Pipeline
Trasferire stringhe del tipo #k* (k = 1 .. 5) da un produttore ad un consumatore
41
THREAD
Pipeline
public class Trasmettitore extends Thread {
private PipedWriter out = new PipedWriter();
private char[] bufChar = new char[3];
public PipedWriter getPipedWriter() { return out; }
public void run() {
for (char c = '1'; c <= '5'; c++) {
try {
bufChar[0] = '#';
bufChar[1] = c;
out.write (bufChar, 0, 2);
bufChar[2] = '*';
out.write (bufChar, 2, 1);
sleep (100);
} catch (Exception e) { }
}
}
}
// Scrive i primi 2 caratteri sulla pipe
// Scrive il 3 carattere sulla pipe
42
THREAD
Pipeline
public class Ricevitore extends Thread {
private PipedReader in;
private char[] bufChar = new char[3];
public Ricevitore (Trasmettitore mittente) throws IOException {
in = new PipedReader (mittente.getPipedWriter());
}
public void run() {
try {
while (true) {
in.read (bufChar, 0, 3);
System.out.println ("Stringa trasferita: "
+ bufChar[0] + bufChar[1] + bufChar[2]);
}
} catch (IOException e) { }
}
}
43
THREAD
public class Pgm_xPipeLine {
public static void main (String[] args) throws Exception {
Trasmettitore mittente = new Trasmettitore();
Ricevitore destinatario = new Ricevitore (mittente);
mittente.start();
destinatario.start();
}
}
Pipeline

Download Report