Marco Faella » 21.Condition variable. Code bloccanti

Le condition variable

Le condition variable (variabili di condizione) sono un classico meccanismo di sincronizzazione, che consente ad un thread di attendere una condizione arbitraria, che altri thread renderanno vera.
Supponiamo due thread condividano una variabile intera e che il primo thread debba aspettare che il secondo ne modifichi il valore per poter andare avanti.
La soluzione “ingenua” consiste nell’utilizzare una ulteriore variabile condivisa, di tipo booleano, e un ciclo del tipo

while (!modified) { /* ciclo vuoto */ }

Questa soluzione prende il nome di attesa attiva, perché durante l’attesa il thread occupa inutilmente la CPU, controllando costantemente la condizione di uscita dal ciclo.
La soluzione corretta, invece, consiste nel realizzare attesa passiva, utilizzando una condition variable.
Una condition variable è un meccanismo di sincronizzazione che, unito ad un mutex, permette di attendere il verificarsi di una condizione tramite attesa passiva e senza rischi di race condition.
Si consulti un testo di sistemi operativi per ulteriori dettagli.
In questa lezione, le condition variable saranno presentate nella loro versione Java.

Le condition variable in Java

Come i mutex, anche le condition variable sono realizzate in Java in modo implicito, ovvero come parte integrante del linguaggio.
In particolare, la loro funzionalità viene offerta dai seguenti metodi della classe Object.

public void wait() throws InterruptedException
public void notify()
public void notifyAll()

Intuitivamente, il metodo wait, chiamato su un oggetto “x”, mette il thread corrente in attesa che qualche altro thread chiami notify o notifyAll sullo stesso oggetto “x”.
Quindi, l’oggetto “x” funge da tramite per permettere al secondo thread di comunicare al primo che può andare avanti nelle sue operazioni.

Come tutti i metodi bloccanti, wait è sensibile allo stato di interruzione del thread, e in caso di interruzione solleva l’eccezione verificata I.E.

La differenza tra notify e notifyAll è che il primo risveglia uno solo dei thread che sono potenzialmente in attesa della condizione, mentre notifyAll li sveglia tutti.

Le prossime slide chiariscono e approfondiscono questi concetti.

Funzionamento interno di wait

Tutti e tre i metodi in esame possono essere chiamati su un oggetto “x” solo se il thread corrente detiene il mutex di “x”; in caso contrario, i metodi lanceranno un’eccezione a runtime.

Vediamo passo per passo il funzionamento interno di una chiamata del tipo “x.wait()“:

1. Se il thread corrente non possiede il mutex di x, lancia un’eccezione;
2. Se lo stato di interruzione del thread corrente è vero, lancia un’eccezione;
3. In un’unica operazione atomica:

• mette il thread corrente nella lista di attesa di x;
• rilascia il mutex di x;
• sospende l’esecuzione del thread.

[i seguenti passi vengono svolti quando il thread viene risvegliato con notify o notifyAll]

4. Riacquisisce il mutex di x;
5. Restituisce il controllo al chiamante.

[se invece il thread viene interrotto con il metodo interrupt durante l'attesa, viene lanciata l'eccezione I.E.].

Osservazioni e il funzionamento di notify(All)

Riguardo la slide precedente, si osservi che al passo 4 il thread che ha invocato wait tenta di riacquisire il mutex dell’oggetto “x”, ma non è detto che ci riesca subito.
Quindi, il thread resta bloccato al passo 4 finché il mutex non si rende disponibile.

Vediamo ora passo per passo il funzionamento interno di una chiamata del tipo “x.notify()” oppure “x.notifyAll()”:

1. Se il thread corrente non possiede il mutex di x, lancia un’eccezione.

2. Se si tratta di notify: preleva un thread dalla coda di attesa di “x” e lo rende nuovamente eseguibile (informalmente, diremo che lo “sveglia”).

Se invece si tratta di notifyAll: preleva tutti i thread dalla coda di attesa di “x” e li rende nuovamente eseguibili.

3. Restituisce il controllo al chiamante.

Applicazione delle condition variable

Una situazione ricorrente nella programmazione concorrente consiste nel paradigma produttore-consumatore.
Si tratta di due o più thread, divisi in due categorie:

1. I produttori (producer) sono fonti di informazioni destinate ai consumatori;
2. I consumatori (consumer) devono elaborare le informazioni fornite dai produttori, non appena queste si rendono disponibili.

Non è possibile prevedere quanto tempo impiega un produttore a produrre un’informazione, né quanto impiega un consumatore ad elaborarla.
Quindi, si pone il problema di sincronizzare le operazioni tra le due categorie.

La soluzione classica prevede uno o più buffer, che contengono le informazioni prodotte e non ancora consumate.
Supponiamo che ci sia un unico buffer, con una capienza limitata.
Se un produttore trova il buffer pieno quando è pronto a produrre una nuova informazione, deve attendere che un consumatore liberi un posto nel buffer.
Simmetricamente, se un consumatore trova il buffer vuoto, deve attendere che un produttore vi inserisca almeno un’informazione.
Le condition variable permettono di realizzare queste attese in modo passivo e senza rischio di race condition.

Nella prossima slide, vediamo lo schema di un’implementazione in Java di questa situazione.

Produttore-consumatore in Java

Supponiamo che il riferimento “buffer” punti ad una struttura dati con metodi “add”, per aggiungere un elemento, e “remove”, per rimuoverne uno. I due thread seguenti usano il buffer non solo per comunicare, ma anche per sincronizzarsi.

Produttore:

synchronized (buffer) {


// attende che il buffer non sia pieno
while (buffer.isFull()) {


try {


buffer.wait();


} catch (InterruptedException e) {


return;


}


}
buffer.add(some_value);
// notifica i consumatori
buffer.notifyAll();


}

Produttore-consumatore in Java (segue)

Consumatore:

synchronized (buffer) {


// attende che il buffer non sia vuoto
while (buffer.isEmpty()) {


try {


buffer.wait();


} catch (InterruptedException e) {


return;


}


}
some_value = buffer.remove();
// notifica i produttori
buffer.notifyAll();


}

Osservazioni

Viene naturale chiedersi perché sia il produttore sia il consumatore organizzano basano la loro attesa su di un ciclo “while”, invece di un semplice “if”.
Cosa ci sarebbe di sbagliato se il produttore fosse strutturato come segue?

if (buffer.isFull()) {


try {


wait();


} catch (InterruptedException e) {


return;
}


}

Il problema che può verifcarsi è il seguente. Supponiamo che più produttori utilizzino lo stesso buffer, e che quest’ultimo sia momentaneamente pieno. Quando un consumatore rimuoverà un’informazione dal buffer, chiamerà notifyAll, risvegliando tutti i produttori che erano in attesa. Il produttore che avrà la fortuna di ottenere per primo il mutex che gli serve per proseguire produrrà una nuova informazione, rendendo nuovamente pieno il buffer. A questo punto, tutti gli altri produttori, che sono stati svegliati, usciranno dal blocco “if”, assumendo erroneamente che il buffer non sia più pieno.

Le interfacce Queue e BlockingQueue

L’interfaccia Queue aggiunge alcuni metodi a Collection, tra i quali menzioniamo solamente peek:

public E peek()

Il metodo peek restituisce il primo elemento della coda, senza rimuoverlo.
Se la coda è vuota, viene restituito null
Questo metodo non è bloccante.

Siccome Queue estende Collection, tutte le code dispongono dei metodi offerti da Collection, quali add, contains e remove.
Inoltre, Queue estende indirettamente Iterable.
Si noti che LinkedList implementa, oltre a List, anche Queue, ma non BlockingQueue.

L’interfaccia parametrica BlockingQueue rappresenta una generica coda bloccante.
Essa offre dei metodi per inserire e rimuovere elementi, che si bloccano se la coda è piena o vuota, rispettivamente.
Così facendo, essa permette a produttori e consumatori di sincronizzarsi senza usare esplicitamente le apposite primitive (mutex e condition variable).
Nella prossima slide, esamineremo i due metodi principali di BlockingQueue.

Metodi delle code bloccanti

L’interfaccia BlockingQueue offre, tra gli altri, i metodi put e take:

public void put(E elem) throws InterruptedException

Inserisce l’oggetto elem in cima alla coda.
Se la coda è piena, questo metodo mette il thread corrente in attesa che venga rimosso qualche elemento dalla coda.
Come tutti i metodi bloccanti, put è sensibile allo stato di interruzione del thread corrente, e solleva l’eccezione I.E. se tale stato diventa vero durante l’attesa, o era già vero all’inizio dell’attesa.

public E take() throws InterruptedException

Restituisce l’elemento in cima alla coda.
Se la coda è vuota, questo metodo mette in attesa in thread corrente, finché non viene inserito almeno un elemento nella coda.
Vale per take lo stesso discorso fatto per put, in relazione allo stato di interruzione dei thread.

Implementazioni di BlockingQueue

La classe ArrayBlockingQueue (in breve, ABQ) è un’implementazione di BlockingQueue, realizzata internamente tramite un array circolare.
Una ABQ ha una capacità fissa, dichiarata una volta per tutte al costruttore:

public ArrayBlockingQueue(int capacity)

Una ABQ rappresenta un classico buffer con capacità limitata (bounded buffer).

La classe LinkedBlockingQueue (in breve, LBQ), invece, è una coda bloccante realizzata tramite una lista concatenata.
Una LBQ ha una capacità potenzialmente illimitata.
Quindi, una LBQ non risulta mai piena.
Pertanto, il metodo put di LBQ non risulta bloccante.

Oltre ad essere bloccanti, queste classi sono anche thread-safe.
Ovvero, sono predisposte all’accesso concorrente da parte di più thread.
Lo stesso non si può dire per le altre classi del Java Collection Framework, come LinkedList o HashSet, che possono dare risultati inattesi se utilizzate concorrentemente da più thread senza le opportune precauzioni di sincronizzazione.

Esercizio (esame 14/9/2010)

Implementare il metodo statico executeInParallel, che accetta come argomenti un array di oggetti di tipo Runnable e un numero naturale k, ed esegue tutti i Runnable dell’array, k alla volta.

In altre parole, all’inizio il metodo fa partire in parallelo i primi k Runnable dell’array.
Poi, non appena uno dei Runnable in esecuzione termina, il metodo ne fa partire un altro, preso dall’array, fino ad esaurire tutto l’array.

Risolvere l’esercizio senza utilizzare attesa attiva.