Monitor e variabili condizione

1. Che cos'è un monitor

I semafori sono di basso livello: la politica di sincronizzazione è sparsa nel codice dei processi (capitolo 5). Il monitor, introdotto da Brinch Hansen (1973) e generalizzato da Hoare (1974), è un'astrazione di dati che incapsula la politica di sincronizzazione e di mutua esclusione nell'accesso ai dati stessi: stato, operazioni e politica di concorrenza in un unico modulo. Generalizza due idee note:

• il kernel del sistema operativo, che centralizza le sezioni critiche (per esempio l'allocazione della memoria) in un programma privilegiato: un monitor è un kernel decentralizzato, uno per risorsa;
• l'oggetto dell'OOP, a cui aggiunge la politica di sincronizzazione.

monitor Counter {                 -- monitor MonitorName {
  int count                       --   variabili permanenti
                                  --   inizializzazione
  procedure inc() {               --   procedure (entry)
    count := count + 1            -- }
  }
  procedure getValue(): int {
    return count
  }
}

Le proprietà di un monitor sono tre:

• dall'esterno sono visibili solo i nomi delle procedure: sono l'interfaccia, le sole porte nel «muro» del monitor, e si chiamano con Counter.inc();
• le istruzioni del monitor non possono accedere a variabili dichiarate fuori dal monitor;
• le variabili permanenti sono inizializzate prima che qualsiasi procedura venga chiamata.

A queste si aggiunge la mutua esclusione implicita: al più una procedura del monitor è attiva in ogni istante, e chi trova il monitor occupato viene sospeso. Il contatore è thread-safe senza nessuna wait o signal: le operazioni di uno stesso monitor sono atomiche l'una rispetto all'altra, mentre quelle di monitor diversi possono alternarsi. Non c'è una coda esplicita per chi aspetta di entrare, quindi la starvation resta possibile.

Nella progettazione il monitor è il componente passivo che i thread, componenti attivi, usano per coordinarsi, ed è ciò che chiede il primo assignment (capitolo 8).

2. Aspettare una condizione: le variabili condizione

La mutua esclusione non basta: un consumatore deve poter aspettare che il buffer non sia vuoto. Una variabile condizione, usabile solo dentro il monitor, rappresenta una condizione sullo stato: serve a sospendere un processo finché la condizione non diventa vera e a risvegliarlo quando lo diventa. Ha una coda FIFO di processi bloccati e due operazioni atomiche:

waitC(cond)   = < append p to cond.queue
                  p.state := blocked
                  monitor.lock := release >    -- rilascia il monitor

signalC(cond) = < if cond.queue != empty
                    q := remove head of cond.queue
                    q.state := ready >

Altre primitive: emptyC(cond) dice se la coda è vuota; signalAllC(cond) risveglia tutti i processi in coda; waitC(cond, rank) accoda in ordine di rank crescente; minrank(cond) restituisce il rank del primo della coda.

Il legame tra la variabile condizione e il suo monitor è essenziale: waitC rilascia il lock del monitor nello stesso passo atomico in cui sospende il processo. Se non lo rilasciasse, chi aspetta terrebbe il monitor occupato e nessuno potrebbe entrare a rendere vera la condizione: deadlock. Se lo rilasciasse in un passo separato, una signalC arrivata nell'intervallo troverebbe la coda vuota e andrebbe persa. Per questo in Java una Condition si ottiene dal proprio lock con lock.newCondition(), e wait() si chiama solo tenendo il lock dell'oggetto.

L'esempio delle slide è una cella sincronizzata: get aspetta che un valore sia stato scritto.

monitor SynchCell {
  int value
  boolean available := false
  cond isAvail

  procedure set(int v) {
    value := v
    available := true
    signalC(isAvail)
  }
  procedure get(): int {
    if (!available)
      waitC(isAvail)
    return value
  }
}

Con più lettori in attesa, signalC ne sveglia uno solo e gli altri restano bloccati anche se il valore è disponibile. Ci sono due rimedi. set può usare signalAllC(isAvail). Oppure ogni lettore risvegliato, appena ripreso, esegue a sua volta signalC(isAvail) e sveglia il successivo: è una segnalazione a cascata.

3. Chi riparte dopo una signal? Le discipline di segnalazione

Dopo una signalC due processi potrebbero eseguire nel monitor, chi ha segnalato e chi è stato risvegliato, ma ne può proseguire uno solo. La disciplina di segnalazione decide quale, e con che precedenza rispetto a chi aspetta di entrare. Nelle slide S è la precedenza di chi ha segnalato, W di chi aspettava sulla condizione, E di chi è bloccato all'ingresso.

La conseguenza pratica riguarda il test prima di waitC. Con Signal and Wait o Signal and Urgent Wait il risvegliato riparte subito: se si segnala solo a condizione vera, la condizione vale ancora e basta un if. Con Signal and Continue, tra la signalC e la ripresa il segnalante continua e possono entrare altri processi, che possono renderla di nuovo falsa: va ricontrollata.

if (!B)  waitC(cond)      -- basta con signal-and-wait / signal-and-urgent-wait
while (!B) waitC(cond)    -- necessario con signal-and-continue (e in Java)
<accesso>

Il while è corretto con qualsiasi disciplina: nel caso peggiore costa un controllo in più.

4. Buffer e lettori-scrittori come monitor

Con un monitor si realizza qualsiasi risorsa condivisa incapsulandone la politica. Le slide rivedono così i problemi del capitolo 5 e aggiungono l'allocatore di risorse: request blocca il richiedente finché la risorsa non è disponibile secondo una politica, release sblocca le richieste in attesa.

Il buffer limitato

monitor BoundedBuffer {
  int[] elems := new int[MAX_ELEMS]
  int first := 0, last := 0
  cond notFull, notEmpty

  procedure put(int elem) {
    if ((last + 1) % MAX_ELEMS = first)   -- pieno
      waitC(notFull)
    elems[last] := elem
    last := (last + 1) % MAX_ELEMS
    signalC(notEmpty)
  }
  procedure take(): int {
    if (first = last)                     -- vuoto
      waitC(notEmpty)
    int elem := elems[first]
    first := (first + 1) % MAX_ELEMS
    signalC(notFull)
    return elem
  }
}

-- producer:  loop  el := produce; BoundedBuffer.put(el)
-- consumer:  loop  el := BoundedBuffer.take(); consume(el)

I processi si limitano a chiamare put e take: mutua esclusione e sincronizzazione stanno nel monitor. Con questo test di «pieno» l'array usa MAX_ELEMS − 1 posti, per distinguere pieno da vuoto. L'if presuppone la ripresa immediata; in Java diventa while (sezione 5).

Lettori e scrittori

Il monitor delle slide è pensato per Signal and Continue, quindi usa while. Il suo invariante è (nr = 0 ∨ nw = 0) ∧ nw ≤ 1.

monitor RWLock {
  int nr := 0, nw := 0
  cond okToRead, okToWrite

  procedure request_read() {
    while (nw > 0) waitC(okToRead)
    nr := nr + 1
  }
  procedure release_read() {
    nr := nr - 1
    if (nr = 0) signalC(okToWrite)
  }
  procedure request_write() {
    while (nr > 0 or nw > 0) waitC(okToWrite)
    nw := nw + 1
  }
  procedure release_write() {
    nw := nw - 1
    signalC(okToWrite)
    signalAllC(okToRead)
  }
}

-- reader:  RWLock.request_read();  read the dbase;  RWLock.release_read()
-- writer:  RWLock.request_write(); write the dbase; RWLock.release_write()

Il monitor non offre read e write, ma operazioni che aprono e chiudono l'accesso. Se la lettura fosse una procedura del monitor, la mutua esclusione implicita impedirebbe a due lettori di leggere insieme: si tornerebbe alla soluzione sovravincolata del capitolo 5, con il monitor occupato per tutta la durata di letture e scritture. Il monitor gestisce quindi solo il protocollo di accesso, i permessi contati da nr e nw, come un allocatore di risorse. L'accesso vero avviene fuori, tra le due chiamate, e ogni processo deve ricordarsi di rilasciare il permesso. ReentrantReadWriteLock segue lo stesso schema (capitolo 10).

La seconda soluzione delle slide (Ben-Ari) è scritta per la ripresa immediata, e usa if e le code delle condizioni per decidere a chi tocca:

monitor RWLock {
  integer readers := 0, writers := 0
  cond okToRead, okToWrite

  procedure startRead() {
    if writers != 0
      waitC(okToRead)
    readers := readers + 1
    signalC(okToRead)          -- a cascata: sveglia il prossimo lettore
  }
  procedure endRead() {
    readers := readers - 1
    if readers = 0
      signalC(okToWrite)
  }
  procedure startWrite() {
    if writers != 0 or readers != 0
      waitC(okToWrite)
    writers := writers + 1
  }
  procedure endWrite() {
    writers := writers - 1
    if emptyC(okToRead)
      then signalC(okToWrite)
      else signalC(okToRead)   -- i lettori in attesa hanno la precedenza
  }
}

5. Monitor in Java

Java offre due modi per scrivere un monitor: i meccanismi di base del linguaggio (synchronized e i metodi wait, notify, notifyAll di Object) oppure ReentrantLock e Condition di java.util.concurrent. Non vanno mescolati nella stessa classe.

Primo modo: synchronized, wait, notify

Ogni oggetto Java ha un lock intrinseco, detto monitor lock (le istruzioni bytecode di un blocco synchronized sono monitorenter e monitorexit). Un oggetto segue il monitor pattern se incapsula tutto il suo stato mutabile e lo protegge con il proprio lock:

• ogni metodo pubblico è synchronized;
• non ci sono campi pubblici;
• il codice del monitor usa solo oggetti completamente confinati al suo interno;
• c'è una sola variabile condizione, implicita: l'oggetto stesso. wait fa da waitC, notify e notifyAll da signalC e signalAllC.

Per ogni oggetto la JVM tiene l'entry set, i thread in attesa del lock, e il wait set, quelli che hanno eseguito wait. Sono insiemi, non code: nessun ordine garantito.

Si chiamano solo tenendo il lock dell'oggetto, altrimenti si ottiene IllegalMonitorStateException. Un interrupt() su un thread in wait agisce come una notifica, ma dopo aver riacquisito il lock il thread esce con InterruptedException e il flag di interruzione torna falso. È il caso in cui wait, come sleep e join, lancia InterruptedException. Un thread bloccato all'ingresso di un blocco synchronized, invece, non si può interrompere.

Perché while e notifyAll

La semantica di Java, con synchronized come con ReentrantLock, è una variante di Signal and Continue con E = W < S: chi notifica continua, e il risvegliato compete alla pari con chi vuole entrare. Nell'esperimento delle lab notes T1 entra ed esegue await; T2 entra, dorme cinque secondi, esegue signal ed esce; nel frattempo T3 si è bloccato all'ingresso.

First thread inside, going to wait
Second thread inside
Third thread started.
Second thread inside, going to signal
Second thread inside signaled.
Third thread inside.          <- T3 entra prima di T1, già risvegliato
First thread unblocked.

In più il meccanismo di base ha una sola variabile condizione, quindi produttori e consumatori aspettano nello stesso wait set, e la specifica ammette i risvegli spuri: «un thread può svegliarsi anche senza essere stato notificato, interrotto o senza timeout». Lo schema sicuro è quindi un while sulla condizione specifica e notifyAll. Con notify si può svegliare il thread sbagliato. Buffer da un posto, due consumatori in attesa: il produttore inserisce, sveglia C1, riprova a inserire e aspetta perché il buffer è pieno. C1 preleva e con notify sveglia C2 invece del produttore; C2 trova il buffer vuoto e torna ad aspettare, C1 al giro dopo pure. Ora tutti e tre aspettano, e nessuno notificherà più.

public final class MonitorBuffer<T> {
    private final Object[] buffer;
    private int first, last, count;

    public MonitorBuffer(int capacity) {
        buffer = new Object[capacity];
    }

    public synchronized void put(T item) throws InterruptedException {
        while (count == buffer.length) {
            wait();
        }
        buffer[last] = item;
        last = (last + 1) % buffer.length;
        count++;
        notifyAll();
    }

    @SuppressWarnings("unchecked")
    public synchronized T get() throws InterruptedException {
        while (count == 0) {
            wait();
        }
        T item = (T) buffer[first];
        buffer[first] = null;
        first = (first + 1) % buffer.length;
        count--;
        notifyAll();
        return item;
    }
}

public final class ConditionBuffer<T> {
    private final Object[] buffer;
    private int first, last, count;
    private final ReentrantLock mutex = new ReentrantLock();
    private final Condition notFull = mutex.newCondition();
    private final Condition notEmpty = mutex.newCondition();

    public ConditionBuffer(int capacity) {
        buffer = new Object[capacity];
    }

    public void put(T item) throws InterruptedException {
        mutex.lockInterruptibly();
        try {
            while (count == buffer.length) {
                notFull.await();
            }
            buffer[last] = item;
            last = (last + 1) % buffer.length;
            count++;
            notEmpty.signal();
        } finally {
            mutex.unlock();
        }
    }

    @SuppressWarnings("unchecked")
    public T get() throws InterruptedException {
        mutex.lockInterruptibly();
        try {
            while (count == 0) {
                notEmpty.await();
            }
            T item = (T) buffer[first];
            buffer[first] = null;
            first = (first + 1) % buffer.length;
            count--;
            notFull.signal();
            return item;
        } finally {
            mutex.unlock();
        }
    }
}

Secondo modo: ReentrantLock e Condition

Un ReentrantLock per monitor, e il corpo di ogni metodo pubblico tra lock() e unlock(), con unlock() in un finally; niente synchronized. Le condizioni si creano dal lock con newCondition(), quante se ne vuole, e await, signal, signalAll fanno da waitC, signalC, signalAllC. Il vantaggio è la segnalazione mirata: put sveglia solo i consumatori, get solo i produttori. La disciplina resta E = W < S, e il while resta obbligatorio.

• Rientrante vuol dire che il thread che possiede il lock può riacquisirlo senza bloccarsi: il lock conta le acquisizioni (hold count) e si libera dopo altrettanti unlock(). Anche synchronized è rientrante; i semafori no.
• L'equità è un'opzione separata: new ReentrantLock(true) favorisce chi aspetta da più tempo, il default non garantisce alcun ordine.
• lockInterruptibly() e tryLock() permettono di rinunciare all'attesa del lock, cosa impossibile con synchronized (capitolo 6).
• Errore frequente: chiamare wait() o notifyAll() su una Condition. Compila, perché sono metodi di Object, ma a runtime lancia IllegalMonitorStateException; i metodi giusti sono await() e signalAll().

Chiamare un monitor da un altro monitor

Una procedura che chiama un altro monitor tiene il proprio lock per tutta la chiamata: se il secondo monitor è occupato, anche il primo resta bloccato per tutti. Se un altro thread fa il percorso inverso si ha il deadlock di ordinamento del capitolo 6, come nel pattern Observer. E se la procedura interna esegue wait, rilascia solo il lock interno: nessuno può entrare nel monitor esterno a rendere vera la condizione (nested monitor lockout). Le chiamate ad altri monitor da dentro un monitor vanno quindi evitate o tenute sotto stretto controllo.

6. Monitor e semafori

Monitor e semafori hanno la stessa potenza espressiva: ciascuno si realizza con l'altro. Un semaforo come monitor, nelle due versioni delle slide:

monitor Semaphore {                  monitor Semaphore {
  integer s := <InitValue>             integer s := <InitValue>
  cond notZero                         cond notZero
  procedure wait() {                   procedure wait() {
    if s = 0                             if s = 0
      waitC(notZero)                       waitC(notZero)   -- riceve il permesso
    s := s - 1                           else
  }                                        s := s - 1
  procedure signal() {                 }
    s := s + 1                         procedure signal() {
    signalC(notZero)                     if emptyC(notZero)
  }                                        s := s + 1
}                                        else
                                           signalC(notZero) -- passa il permesso
                                       }
                                     }

La prima versione usa if, quindi presuppone la ripresa immediata; con Signal and Continue serve while s = 0. La seconda passa il permesso direttamente al processo risvegliato, esattamente come la signal del semaforo.

-- Signal and Continue                 -- Signal and Wait
waitC(cond) =                          waitC(cond) =
  condcount++                            condcount++
  signal(mutex)                          signal(mutex)
  wait(condsem)                          wait(condsem)   -- eredita il mutex
  wait(mutex)                          signalC(cond) =
signalC(cond) =                          if condcount > 0
  if condcount > 0                         condcount--
    condcount--                            signal(condsem)
    signal(condsem)                        wait(mutex)   -- il segnalante aspetta

A parità di potenza i monitor sono preferibili perché di livello più alto: la mutua esclusione è implicita e non si può dimenticare un rilascio, la politica sta in un solo punto dentro la risorsa invece che sparsa nei processi, e l'incapsulamento avvicina l'implementazione al design. Le variabili condizione però non sono semafori:

Per questo una variabile condizione va sempre accompagnata da una condizione esplicita sullo stato (available, count, …): è lo stato, non la variabile, a ricordare che cosa è successo.

7. Barriere e latch: componenti di coordinazione

Con i monitor si costruiscono componenti di coordinazione riutilizzabili; le slide elencano latch, barriere, rendez-vous, message box, blackboard ed event service. In laboratorio si sono scritti la barriera e il latch.

Una barriera è un punto di sincronizzazione simmetrico. N processi eseguono una prima fase, e nessuno comincia la seconda finché tutti non hanno finito la prima. Il monitor deve contare gli arrivati: chi arriva e non trova tutti aspetta, e l'ultimo sveglia tutti. Un latch, o gate, è un cancello con due ruoli distinti. Chi chiama await aspetta che il cancello si apra. I processi che passano chiamano countDown senza mai bloccarsi, e dopo N countDown il cancello si apre e resta aperto per sempre. In breve, le barriere servono ad aspettare altri thread, i latch ad aspettare eventi.

public final class Barrier {
    private final int nParticipants;
    private int nArrived = 0;
    private final ReentrantLock lock = new ReentrantLock();
    private final Condition allArrived = lock.newCondition();

    public Barrier(int nParticipants) { this.nParticipants = nParticipants; }

    public void hitAndWaitAll() throws InterruptedException {
        lock.lock();
        try {
            nArrived++;
            if (nArrived == nParticipants) {
                allArrived.signalAll();         // l'ultimo sveglia tutti
            }
            while (nArrived < nParticipants) {
                allArrived.await();
            }
        } finally {
            lock.unlock();
        }
    }
}

public final class Latch {
    private final int nCountDowns;
    private int nCounts = 0;

    public Latch(int nCountDowns) { this.nCountDowns = nCountDowns; }

    public synchronized void countDown() {      // chi passa: non si blocca
        nCounts++;
        if (nCounts == nCountDowns) {
            notifyAll();                        // il cancello si apre
        }
    }

    public synchronized void await() throws InterruptedException {
        while (nCounts < nCountDowns) {
            wait();
        }
    }
}

Nel latch si notifica solo quando il conteggio arriva a N. Notificare a ogni countDown sarebbe corretto, grazie al while, ma sveglierebbe chi aspetta solo per farlo riaddormentare. La barriera così scritta si usa una volta sola. Renderla ciclica non è banale. Azzerare nArrived quando arrivano tutti non basta, perché un thread veloce può rientrare nel giro successivo prima che quelli del giro precedente si siano risvegliati. Servono le «generazioni» di java.util.concurrent.CyclicBarrier, che con CountDownLatch è la versione di libreria di questi componenti (capitolo 10).

Verifica le tue conoscenze