Semafori, monitor e problemi classici di sincronizzazione

Il mutex di Sincronizzazione - sezione critica, race condition e mutexQuando più processi o thread accedono agli stessi dati e il risultato dipende dall'ordine degli accessi si ha una race condition (esempio: counter++ e counter-- sono tre istruzioni macchina ciascuna, e interleaving diversi danno 4, 5 o 6 partendo da 5); il problema della sezione critica chiede un protocollo (entry section, sezione critica, exit section, remainder section) che garantisca mutua esclusione, progresso e attesa limitata; disabilitare gli interrupt non scala sui multiprocessori, l'alternanza stretta con turn viola il progresso, la soluzione di Peterson (flag[2] e turn) soddisfa i tre requisiti ma può fallire su hardware moderno per il riordino delle istruzioni, per cui servono barriere di memoria; l'hardware offre test_and_set e compare_and_swap (atomiche) e variabili atomiche; il mutex lock (acquire/release, in POSIX pthread_mutex_t) è lo strumento di alto livello, con attesa attiva se è uno spinlock oppure con sospensione del thread se è un mutex di sistema; bloccare due volte lo stesso mutex senza sbloccarlo causa deadlock.Sincronizzazione - sezione critica, race condition e mutex → risolve la mutua esclusione ma non basta per coordinare i processi: far aspettare uno finché l'altro non ha prodotto un dato, limitare a $n$ i processi in una zona, ordinare le operazioni. Gli strumenti di questa nota sono il semaforo e il monitor; poi si applicano ai tre problemi classici (produttore-consumatore, lettori-scrittori, filosofi a cena) con programmi completi in C e in Go. Gli esercizi d'esame sulla sincronizzazione sono nell'elenco in fondo.

Semafori

Definizione (semaforo). Un semaforo $S$ è una variabile intera che si può usare solo con due operazioni atomiche: wait(S) (storicamente $P$, "proberen") e signal(S) (storicamente $V$, "verhogen"; in POSIX sem_post).

wait(S) {                     signal(S) {
    while (S <= 0) ;              S++;
    S--;                      }
}

wait aspetta che il valore sia positivo e poi lo decrementa; signal lo incrementa. Il valore non scende mai sotto zero nell'uso POSIX. Si distinguono:

• semaforo contatore (counting semaphore): il valore può essere un intero qualunque non negativo, per esempio il numero di risorse libere;
• semaforo binario (binary semaphore): vale 0 o 1. È simile a un mutex ma più flessibile: può essere bloccato da un thread e sbloccato da un altro; un mutex deve essere bloccato e sbloccato dallo stesso thread.

Due usi tipici

1. Mutua esclusione: si crea mutex inizializzato a 1: wait(mutex); sezione critica; signal(mutex);.
2. Imporre un ordine: $P_1$ esegue $S_1$ e $P_2$ esegue $S_2$, ma $S_1$ deve precedere $S_2$. Si crea synch inizializzato a 0:

P1:  S1;  signal(synch);        P2:  wait(synch);  S2;

$P_2$ si blocca sulla wait finché $P_1$ non ha finito $S_1$ e fatto signal. Con più semafori si realizza qualunque grafo di precedenza (Esercizio - Grafi di precedenza e ordine di stampa con i semafori (eserciziario, esercizi 19, 52, 60, 91 e 114)).

Semafori senza attesa attiva

La versione sopra usa attesa attiva. Per evitarla, a ogni semaforo si associa una coda di attesa: la struttura ha il valore e un puntatore alla lista dei processi sospesi. Due operazioni interne: sleep (mette il processo in attesa nella coda) e wakeup (toglie un processo dalla coda e lo mette nella ready queue).

wait(S) {                          signal(S) {
    S.value--;                         S.value++;
    if (S.value < 0) {                 if (S.value <= 0) {
        aggiungi questo processo           togli un processo P da S.list;
            a S.list;                      wakeup(P);
        sleep();                       }
    }                              }
}

In questa definizione il valore può diventare negativo, e $|\text{valore}|$ è il numero di processi sospesi (nelle implementazioni reali, come POSIX, si tiene il valore $\ge 0$ e la coda a parte). Anche qui wait e signal devono essere atomiche: si realizzano con compare_and_swap o test_and_set sul campo del semaforo; resta un'attesa attiva, ma molto breve (solo la durata di wait e signal).

Semafori POSIX

I semafori con nome vivono in /dev/shm: se un programma viene interrotto con Ctrl-C senza sem_unlink, al riavvio trova i semafori vecchi con valori sbagliati; si rimuovono a mano con rm /dev/shm/sem.nome. Si compila con -lpthread.

I rischi dei semafori

L'uso scorretto produce errori subdoli: signal(mutex) prima di wait(mutex) (due processi nella sezione critica), wait(mutex) due volte di fila (blocco permanente), omissione di wait o di signal. Sono la ragione per cui si introducono i monitor.

Monitor

Definizione (monitor). Un monitor è un tipo di dato astratto che incapsula dati condivisi e procedure per usarli; le variabili interne sono accessibili solo dal codice delle procedure, e solo un processo alla volta può essere attivo dentro il monitor. La mutua esclusione è garantita dalla costruzione.

monitor nome {
    // dichiarazioni delle variabili condivise
    procedure F1(...) { ... }
    procedure F2(...) { ... }
    codice di inizializzazione(...) { ... }
}

Un monitor si può realizzare con un solo semaforo mutex = 1: ogni procedura $F$ diventa wait(mutex); corpo di F; signal(mutex);.

Variabili condizione

Per aspettare una condizione dentro il monitor si usano le variabili condizione: condition x; con due operazioni:

• x.wait(): il processo che la invoca è sospeso (e lascia libero il monitor) finché un altro invoca x.signal();
• x.signal(): risveglia uno dei processi sospesi su x; se non ce n'è nessuno non ha alcun effetto (a differenza di signal sul semaforo, che incrementa il valore: la variabile condizione non ha memoria).

Se signal risveglia $Q$ mentre $P$ è nel monitor, uno dei due deve aspettare: signal and wait ($P$ attende che $Q$ lasci il monitor o si sospenda) oppure signal and continue ($Q$ attende che $P$ lasci il monitor). Nella seconda ($Q$ potrebbe trovare la condizione ormai falsa) la wait si mette sempre in un while, non in un if.

Se più processi sono sospesi su $x$, quale riprendere? FCFS non è sempre adeguato: con la conditional wait x.wait(c) il numero di priorità c decide, il più basso va prima. Esempio: l'allocatore di una risorsa dove $c$ è il tempo massimo di uso previsto: R.acquire(t); ... usa la risorsa ...; R.release();: si serve prima chi la userà meno.

I monitor non sono nell'API POSIX, che offre mutex e variabili condizione (pthread_cond_t: pthread_cond_wait(&cond, &mutex) rilascia il mutex e sospende, pthread_cond_signal risveglia uno, pthread_cond_broadcast tutti). Il produttore-consumatore a buffer limitato in stile monitor:

#include <pthread.h>
#include <stdio.h>

#define N 4
#define PRODOTTI 20

static int buf[N], in, out, conta;                 /* "dati del monitor" */
static pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t non_pieno  = PTHREAD_COND_INITIALIZER;
static pthread_cond_t non_vuoto  = PTHREAD_COND_INITIALIZER;

static void inserisci(int x)                       /* procedura del monitor */
{
    pthread_mutex_lock(&m);
    while (conta == N)                             /* while, non if: la condizione va rivalutata */
        pthread_cond_wait(&non_pieno, &m);         /* rilascia m e sospende */
    buf[in] = x;
    in = (in + 1) % N;
    conta++;
    pthread_cond_signal(&non_vuoto);
    pthread_mutex_unlock(&m);
}

static int preleva(void)                           /* procedura del monitor */
{
    pthread_mutex_lock(&m);
    while (conta == 0)
        pthread_cond_wait(&non_vuoto, &m);
    int x = buf[out];
    out = (out + 1) % N;
    conta--;
    pthread_cond_signal(&non_pieno);
    pthread_mutex_unlock(&m);
    return x;
}

static void *produttore(void *arg)
{
    (void)arg;
    for (int i = 1; i <= PRODOTTI; i++)
        inserisci(i);
    return NULL;
}

int main(void)
{
    pthread_t p;
    pthread_create(&p, NULL, produttore, NULL);
    long somma = 0;
    for (int i = 0; i < PRODOTTI; i++)
        somma += preleva();
    pthread_join(p, NULL);
    printf("somma = %ld (attesa 210)\n", somma);
    return 0;
}

Liveness

La liveness è l'insieme delle proprietà che garantiscono il progresso dei processi; l'attesa indefinita di un lock o di un semaforo viola progresso e attesa limitata. Le forme di fallimento:

Problemi classici

I problemi classici servono a provare nuovi schemi di sincronizzazione.

1. Produttore-consumatore con buffer limitato

Un buffer di $n$ posti. Tre semafori: mutex = 1 (mutua esclusione sul buffer), vuote = n (posti liberi), piene = 0 (posti occupati). Il produttore fa wait(vuote), wait(mutex), inserisce, signal(mutex), signal(piene); il consumatore fa wait(piene), wait(mutex), preleva, signal(mutex), signal(vuote). L'ordine delle wait è importante: se il produttore facesse wait(mutex) prima di wait(vuote), con buffer pieno terrebbe il mutex mentre dorme, e il consumatore non potrebbe mai entrare a prelevare: deadlock.

#include <pthread.h>
#include <semaphore.h>
#include <stdio.h>

#define N 5                    /* posti del buffer */
#define PRODOTTI 20

static int buffer[N];
static int in, out;
static sem_t vuote, piene, mutex;

static void *produttore(void *arg)
{
    (void)arg;
    for (int i = 1; i <= PRODOTTI; i++) {
        sem_wait(&vuote);               /* prima la risorsa... */
        sem_wait(&mutex);               /* ...poi il mutex */
        buffer[in] = i;
        in = (in + 1) % N;
        sem_post(&mutex);
        sem_post(&piene);
    }
    return NULL;
}

static void *consumatore(void *arg)
{
    (void)arg;
    long somma = 0;
    for (int i = 0; i < PRODOTTI; i++) {
        sem_wait(&piene);
        sem_wait(&mutex);
        int x = buffer[out];
        out = (out + 1) % N;
        sem_post(&mutex);
        sem_post(&vuote);
        somma += x;
    }
    printf("somma = %ld (attesa %d)\n", somma, PRODOTTI * (PRODOTTI + 1) / 2);
    return NULL;
}

int main(void)
{
    pthread_t p, c;
    sem_init(&vuote, 0, N);             /* N posti liberi */
    sem_init(&piene, 0, 0);             /* nessun posto occupato */
    sem_init(&mutex, 0, 1);
    pthread_create(&p, NULL, produttore, NULL);
    pthread_create(&c, NULL, consumatore, NULL);
    pthread_join(p, NULL);
    pthread_join(c, NULL);
    sem_destroy(&vuote);
    sem_destroy(&piene);
    sem_destroy(&mutex);
    return 0;
}

In Go il buffer limitato è un canale con capacità: l'invio si blocca se è pieno, la ricezione se è vuoto, e la chiusura segnala la fine. Il canale è, in un solo oggetto, buffer + semafori vuote/piene + mutua esclusione.

package main

import (
	"fmt"
	"sync"
)

const (
	dimBuffer = 5
	prodotti  = 20
)

func produttore(buf chan<- int, wg *sync.WaitGroup) {
	defer wg.Done()
	for i := 1; i <= prodotti; i++ {
		buf <- i // si blocca se il buffer (capacità 5) è pieno
	}
	close(buf) // nessun altro prodotto: il consumatore uscirà dal range
}

func consumatore(buf <-chan int, wg *sync.WaitGroup) {
	defer wg.Done()
	somma := 0
	for x := range buf { // si blocca se il buffer è vuoto, termina alla chiusura
		somma += x
	}
	fmt.Println("somma =", somma, "(attesa 210)")
}

func main() {
	buf := make(chan int, dimBuffer)
	var wg sync.WaitGroup
	wg.Add(2)
	go produttore(buf, &wg)
	go consumatore(buf, &wg)
	wg.Wait()
}

2. Lettori e scrittori

Un insieme di dati condiviso: i lettori leggono soltanto, gli scrittori leggono e scrivono. Si vuole che più lettori leggano insieme, ma che uno scrittore abbia accesso esclusivo. Ci sono varianti: nella prima (priorità ai lettori) nessun lettore attende a meno che uno scrittore sia già dentro; è la soluzione che segue. Dati condivisi: rw_mutex = 1 (accesso esclusivo ai dati), mutex = 1 (protegge il contatore), read_count = 0.

scrittore:                         lettore:
    wait(rw_mutex);                    wait(mutex);
    ... scrive ...                     read_count++;
    signal(rw_mutex);                  if (read_count == 1) wait(rw_mutex);   // il primo blocca gli scrittori
                                       signal(mutex);
                                       ... legge ...
                                       wait(mutex);
                                       read_count--;
                                       if (read_count == 0) signal(rw_mutex); // l'ultimo li libera
                                       signal(mutex);

Perché un semaforo binario e non un mutex per rw_mutex? Perché il primo lettore blocca rw_mutex e l'ultimo (un altro thread) lo sblocca: un mutex deve essere sbloccato dal proprietario. In questa variante gli scrittori possono soffrire di starvation: se continuano ad arrivare lettori mentre uno è dentro, nessuno scrittore entra mai. Nessun deadlock. Altre varianti danno la priorità agli scrittori (e allora i lettori possono morire di fame) o impongono un ordine di arrivo.

#include <pthread.h>
#include <semaphore.h>
#include <stdio.h>
#include <unistd.h>

#define LETTORI 3
#define SCRITTORI 2
#define GIRI 20

static struct { int a, b; } dato;       /* invariante: a == b fuori dalla scrittura */
static sem_t rw_mutex, mutex;
static int read_count;

static void *scrittore(void *arg)
{
    (void)arg;
    for (int k = 0; k < GIRI; k++) {
        sem_wait(&rw_mutex);            /* accesso esclusivo */
        dato.a++;
        usleep(100);                    /* finestra con a != b: nessun lettore deve vederla */
        dato.b = dato.a;
        sem_post(&rw_mutex);
        usleep(50);
    }
    return NULL;
}

static void *lettore(void *arg)
{
    long id = (long)arg;
    for (int k = 0; k < GIRI; k++) {
        sem_wait(&mutex);
        if (++read_count == 1)
            sem_wait(&rw_mutex);        /* il primo lettore blocca gli scrittori */
        sem_post(&mutex);

        if (dato.a != dato.b)
            printf("lettore %ld: dato INCOERENTE\n", id);

        sem_wait(&mutex);
        if (--read_count == 0)
            sem_post(&rw_mutex);        /* l'ultimo li riammette */
        sem_post(&mutex);
        usleep(30);
    }
    return NULL;
}

int main(void)
{
    pthread_t t[LETTORI + SCRITTORI];
    sem_init(&rw_mutex, 0, 1);
    sem_init(&mutex, 0, 1);
    for (long i = 0; i < SCRITTORI; i++)
        pthread_create(&t[i], NULL, scrittore, NULL);
    for (long i = 0; i < LETTORI; i++)
        pthread_create(&t[SCRITTORI + i], NULL, lettore, (void *)i);
    for (int i = 0; i < LETTORI + SCRITTORI; i++)
        pthread_join(t[i], NULL);
    printf("finale: a = %d, b = %d (atteso %d)\n", dato.a, dato.b, SCRITTORI * GIRI);
    return 0;
}

In Go la libreria standard offre sync.RWMutex (RLock/RUnlock per i lettori, Lock/Unlock per gli scrittori), che fa lo stesso lavoro; a differenza della prima variante, quando uno scrittore è in attesa i nuovi lettori si accodano, quindi non c'è starvation degli scrittori.

package main

import (
	"fmt"
	"sync"
	"time"
)

type dati struct {
	a, b int
}

func main() {
	var (
		rw sync.RWMutex
		d  dati
		wg sync.WaitGroup
	)
	for w := 0; w < 2; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for k := 0; k < 50; k++ {
				rw.Lock() // accesso esclusivo
				d.a++
				time.Sleep(10 * time.Microsecond) // finestra con a != b
				d.b = d.a
				rw.Unlock()
			}
		}()
	}
	for r := 0; r < 3; r++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for k := 0; k < 100; k++ {
				rw.RLock() // più lettori insieme
				if d.a != d.b {
					fmt.Println("lettore", id, "ha visto dati incoerenti")
				}
				rw.RUnlock()
			}
		}(r)
	}
	wg.Wait()
	fmt.Println("finale: a =", d.a, "b =", d.b) // 100 100
}

3. I filosofi a cena

$N$ filosofi siedono a un tavolo rotondo con una ciotola di riso al centro e una bacchetta tra ogni coppia (5 bacchette per 5 filosofi). Alternano pensare e mangiare, senza interagire con i vicini; per mangiare servono due bacchette (una alla volta), poi le rilasciano. Dati condivisi: semaphore chopstick[5], tutti a 1.

Soluzione ingenua: wait(chopstick[i]); wait(chopstick[(i+1)%5]); mangia; signal(...); signal(...). Se tutti prendono contemporaneamente la bacchetta sinistra, ognuno aspetta la destra posseduta dal vicino: deadlock.

Rimedi:

La soluzione con monitor tiene per ogni filosofo uno stato THINKING, HUNGRY, EATING e una variabile condizione per ciascuno. pickup(i): stato HUNGRY, test(i), e se non può mangiare self[i].wait(). test(i) mette EATING e fa self[i].signal() se il filosofo $i$ è HUNGRY e i vicini non sono EATING. putdown(i): stato THINKING, poi test sul vicino sinistro e destro. Il filosofo $i$ chiama pickup(i), mangia, chiama putdown(i). Non c'è deadlock, ma è possibile la starvation (due vicini che si alternano possono affamare il filosofo in mezzo).

Soluzione in C con ordinamento delle risorse (la bacchetta di indice minore per prima):

#include <pthread.h>
#include <stdio.h>
#include <unistd.h>

#define N 5
#define PASTI 3

static pthread_mutex_t bacchetta[N];

static void *filosofo(void *arg)
{
    int i = (int)(long)arg;
    int a = i, b = (i + 1) % N;
    if (a > b) {                       /* ordine totale: si prende sempre prima l'indice minore */
        int t = a; a = b; b = t;
    }
    for (int k = 1; k <= PASTI; k++) {
        usleep(100);                   /* pensa */
        pthread_mutex_lock(&bacchetta[a]);
        pthread_mutex_lock(&bacchetta[b]);
        printf("filosofo %d mangia (pasto %d)\n", i, k);
        usleep(100);                   /* mangia */
        pthread_mutex_unlock(&bacchetta[b]);
        pthread_mutex_unlock(&bacchetta[a]);
    }
    return NULL;
}

int main(void)
{
    pthread_t t[N];
    for (int i = 0; i < N; i++)
        pthread_mutex_init(&bacchetta[i], NULL);
    for (long i = 0; i < N; i++)
        pthread_create(&t[i], NULL, filosofo, (void *)i);
    for (int i = 0; i < N; i++)
        pthread_join(t[i], NULL);
    return 0;
}

Perché non va in deadlock: i filosofi $0..3$ prendono prima la bacchetta di sinistra (indice $i$), ma il filosofo 4 prende prima la 0 (indice minore) e poi la 4: non si può chiudere il ciclo di attese, perché ogni attesa va da un indice minore a uno maggiore.

Soluzione in Go con un semaforo (canale con capacità $N-1$) davanti alle bacchette (mutex):

package main

import (
	"fmt"
	"sync"
)

const n = 5

func filosofo(id int, bacchette []sync.Mutex, sedia chan struct{}, wg *sync.WaitGroup) {
	defer wg.Done()
	sx, dx := id, (id+1)%n
	for k := 1; k <= 3; k++ {
		// pensa ...
		sedia <- struct{}{} // "wait" sul semaforo: al massimo n-1 filosofi cercano le bacchette
		bacchette[sx].Lock()
		bacchette[dx].Lock()
		fmt.Printf("filosofo %d mangia (pasto %d)\n", id, k)
		bacchette[dx].Unlock()
		bacchette[sx].Unlock()
		<-sedia // "signal"
	}
}

func main() {
	bacchette := make([]sync.Mutex, n)
	sedia := make(chan struct{}, n-1) // semaforo contatore con valore iniziale n-1
	var wg sync.WaitGroup
	for i := 0; i < n; i++ {
		wg.Add(1)
		go filosofo(i, bacchette, sedia, &wg)
	}
	wg.Wait()
}

Qui un canale con buffer fa da semaforo contatore: l'invio è la wait (si blocca quando il buffer è pieno, cioè quando $N-1$ filosofi sono già in gioco), la ricezione è la signal. Anche questa soluzione evita il deadlock: tra gli $N-1$ filosofi in gioco almeno uno ottiene entrambe le bacchette.

Esercizi collegati

• Esercizio - Output ed ordine di esecuzione con i semafori (eserciziario, esercizi 6, 7 e 8)
• Esercizio - Priorità, semaforo FIFO e code (eserciziario, esercizi 13 e 44)
• Esercizio - Lettori e scrittori, che cosa succede togliendo una istruzione (eserciziario, esercizi 14, 46 e 107)
• Esercizio - Buffer limitato con semafori e messaggi lunghi (eserciziario, esercizi 11 e 45)
• Esercizio - Spogliatoi e cesti della piscina (eserciziario, esercizio 9)
• Esercizio - Bagno condiviso e piazza dei tifosi con due categorie (eserciziario, esercizi 12 e 18)
• Esercizio - Distributore di aranciata, moduli di borsa di studio e attuatori (eserciziario, esercizi 10, 15, 47 e 85)
• Esercizio - Monitor per conto bancario, cannibali e fabbrica (eserciziario, esercizi 16, 17 e 103)
• Esercizio - Il barbiere che dorme con i semafori (eserciziario, esercizio 68)
• Esercizio - Incrocio con il vigile e autobus con due porte (eserciziario, esercizi 75 e 79)
• Esercizio - Grafi di precedenza e ordine di stampa con i semafori (eserciziario, esercizi 19, 52, 60, 91 e 114)
• Esercizio - Semafori POSIX, thread e processi (laboratorio 5)
• Esercizio - Sincronizzazione in Go con canali, mutex e variabili condizione (problemi classici)
• Esercizio - Valori finali di x e y con semafori e processi concorrenti (eserciziario, esercizi 99 e 110)

Domande d'esame

1. Perché in produttore la wait(vuote) va prima di wait(mutex)? Che cosa accade nell'ordine inverso? Traccia. Se il buffer è pieno, un produttore che ha già preso il mutex si sospende su vuote tenendo il mutex: il consumatore, che per liberare un posto deve prendere il mutex, si blocca a sua volta. Nessuno procede: deadlock. Si prende prima la risorsa che può far aspettare, poi il mutex.

2. Differenza tra signal di un semaforo e signal di una variabile condizione. Traccia. Il signal del semaforo incrementa sempre il valore, anche se nessuno aspetta (la chiamata "resta memorizzata"). Il signal della variabile condizione risveglia un processo sospeso, ma se nessuno aspetta non fa nulla: non ha memoria. Per questo la condizione va sempre ricontrollata in un while.

3. Descrivere un deadlock nei filosofi a cena e due modi per evitarlo. Una delle soluzioni può causare starvation? Traccia. Tutti prendono la bacchetta sinistra e aspettano la destra: attesa circolare. Rimedi: al massimo $N-1$ filosofi alla volta, oppure ordine totale sulle bacchette (o asimmetria pari/dispari). La soluzione con monitor evita il deadlock ma può affamare un filosofo i cui due vicini si alternano a mangiare.