Lab 4 — Synchroniser : verrous, sémaphores, files et interblocage

Énoncé du lab 4 d’Architecture et Programmation Parallèles, séance 3 : synchroniser : verrous, sémaphores, files et interblocage.
Auteurs

Dr. El Hadji Bassirou TOURÉ · DMI/FST/UCAD

NotePrésentation

Dès que plusieurs fils partagent une donnée, il faut organiser leurs accès. Ce lab fait voir chaque phénomène de la Séance 3 sur des fils (threading), qui partagent toute la mémoire du notebook : une course qui perd des incréments, le verrou qui la répare, le sémaphore qui limite les accès, la file d’attente qui relie producteurs et consommateurs, le pipeline, et l’interblocage, détecté puis supprimé.

Ici, les fils passent l’essentiel de leur temps à attendre (time.sleep simule une lecture de fichier ou une requête) : c’est le cas où les fils fonctionnent bien en Python, et aucune fonction n’a besoin d’être dans un module.

Prérequis : Labs 0 et 3 (Partie 3). Durée estimée : 1 h 30.

NoteCopier les commandes et le code

Les blocs sans titre sont des commandes à taper dans le terminal (Anaconda Prompt) ; les blocs titrés « Cellule » sont le contenu des cellules du notebook, numérotées dans l’ordre ; les blocs titrés « Résultat » montrent le résultat obtenu (les durées et les numéros de processus changent d’une machine à l’autre). Pour copier un bloc : cliquer sur l’icône de copie, en haut à droite du bloc, puis le coller dans la cellule ; l’indentation est conservée. Le notebook complet est aussi à télécharger (panneau « Pour cette séance » de cette page).

Partie 0 — Mise en place

Cellule 1

import time, threading, queue, random, contextlib, math
from concurrent.futures import ThreadPoolExecutor

def lancer(cible, arguments):
    """un fil par tuple d'arguments ; attend qu'ils finissent tous ; rend la durée"""
    fils = [threading.Thread(target=cible, args=a) for a in arguments]
    t0 = time.perf_counter()
    for f in fils: f.start()
    for f in fils: f.join()
    return time.perf_counter() - t0

print("prêt")

Résultat →

prêt

Partie 1 — Une course critique

compteur = compteur + 1 se fait en trois temps : lire la valeur, ajouter \(1\), écrire le résultat. Si un autre fil lit entre la lecture et l’écriture du premier, l’un des deux incréments est écrasé.

Cellule 2

import dis
compteur = 0
def incrementer_une_fois():
    global compteur
    compteur = compteur + 1
dis.dis(incrementer_une_fois)

Résultat →

  3           RESUME                   0

  5           LOAD_GLOBAL              0 (compteur)
              LOAD_CONST               1 (1)
              BINARY_OP                0 (+)
              STORE_GLOBAL             0 (compteur)
              RETURN_CONST             0 (None)
NoteLecture

LOAD_GLOBAL lit, BINARY_OP ajoute, STORE_GLOBAL écrit : trois instructions, entre lesquelles Python peut passer la main à un autre fil.

Cellule 3

def incrementer_lent(n):
    global compteur
    for _ in range(n):
        v = compteur          # 1. lire
        time.sleep(0)         # le fil cède la main ici, comme pendant une attente
        compteur = v + 1      # 2. ajouter et 3. écrire

compteur = 0
lancer(incrementer_lent, [(1000,), (1000,)])
print("attendu : 2000   obtenu :", compteur)

Résultat →

attendu : 2000   obtenu : 1001
NoteLecture

Environ la moitié des incréments a disparu : presque à chaque tour, les deux fils lisent la même valeur avant que l’un ait écrit. time.sleep(0) rend visible une fenêtre qui existe toujours entre lire et écrire ; sans lui, la course est rare, mais pas impossible.

Partie 2 — Le verrou

Un verrou (threading.Lock) laisse entrer un seul fil à la fois dans le bloc with verrou:. Ce bloc est une section critique.

Cellule 4

verrou = threading.Lock()

def incrementer_protege(n):
    global compteur
    for _ in range(n):
        with verrou:              # prendre le verrou ; il est rendu en sortant
            v = compteur
            time.sleep(0)
            compteur = v + 1

compteur = 0
lancer(incrementer_protege, [(1000,), (1000,)])
print("attendu : 2000   obtenu :", compteur)

Résultat →

attendu : 2000   obtenu : 2000
NoteLecture

Le compte est exact : quand un fil cède la main au milieu, l’autre attend devant with verrou: que le premier ait écrit.

2.1 Vérifier puis agir : la banque

Le compte joint de Fatou et Amadou contient \(10\,000\) F. Chacun, à un guichet différent, retire \(8\,000\) F au même moment. Le guichet vérifie le solde, interroge la banque centrale, puis écrit le nouveau solde.

Cellule 5

def guichet(qui, montant, servis, verrou):
    global solde
    with verrou:                              # contextlib.nullcontext() : aucun verrou
        if solde >= montant:                  # A. vérifier
            nouveau = solde - montant         # B. calculer
            time.sleep(0.01)                  # le guichet interroge la banque centrale
            solde = nouveau                   # C. écrire
            servis.append(qui)

def journee(verrou, nom):
    global solde
    solde, servis = 10_000, []
    lancer(guichet, [("Fatou", 8_000, servis, verrou),
                     ("Amadou", 8_000, servis, verrou)])
    print(f"{nom} : servis {servis}, versé {8_000 * len(servis)} F, solde {solde} F")

journee(contextlib.nullcontext(), "sans verrou")
journee(threading.Lock(), "avec verrou")

Résultat →

sans verrou : servis ['Fatou', 'Amadou'], versé 16000 F, solde 2000 F
avec verrou : servis ['Fatou'], versé 8000 F, solde 2000 F
NoteLecture

Sans verrou, les deux guichets vérifient le solde avant que l’un écrive : la banque verse \(16\,000\) F pour un compte de \(10\,000\) F, et le solde affiche \(2\,000\) F, une valeur qui paraît normale. Avec le verrou, la vérification et l’écriture forment un seul bloc : Amadou trouve \(2\,000\) F et son retrait est refusé.

AstuceÀ faire

Écrire guichet_faux, où la vérification if\ solde\ >=\ montant: est faite avant with verrou: (seuls le calcul et l’écriture sont dans le bloc), et lancer la journée avec un vrai verrou. Que se passe-t-il ?

Cellule 6 — à compléter

# def guichet_faux(qui, montant, servis, verrou): ...

Partie 3 — Le sémaphore : limiter le nombre d’accès

Un sémaphore est un compteur de places. Le parking du Plateau a \(3\) places : une voiture entre s’il reste une place (acquire), et en libère une en sortant (release). Ici, \(10\) voitures arrivent ensemble et stationnent \(0{,}2\) s.

Cellule 7

places = threading.Semaphore(3)
compte = threading.Lock()
dedans, maximum = 0, 0

def voiture(i):
    global dedans, maximum
    with places:                             # attendre une place libre
        with compte:
            dedans += 1
            maximum = max(maximum, dedans)
        time.sleep(0.2)                      # stationnement
        with compte:
            dedans -= 1

duree = lancer(voiture, [(i,) for i in range(10)])
print(f"au plus {maximum} voitures en même temps ; durée {duree:.2f} s")

Résultat →

au plus 3 voitures en même temps ; durée 0.80 s
NoteLecture

Jamais plus de trois voitures. Les voitures entrent par vagues de trois : \(\lceil 10/3\rceil = 4\) vagues de \(0{,}2\) s, soit environ \(0{,}8\) s. Le verrou compte protège les variables dedans et maximum : le sémaphore limite le nombre de fils, il ne protège pas les variables qu’ils partagent.

AstuceÀ faire

Prédire la durée avec \(5\) places, puis avec \(6\) places, en utilisant \(T = \lceil n/k\rceil \times 0{,}2\) s, puis vérifier par la mesure.

Cellule 8 — à compléter

# places = threading.Semaphore(5) ; maximum = 0 ; ...

Partie 4 — Producteurs et consommateurs : la file d’attente

Dans une boulangerie de Kaolack, deux boulangers posent leurs pains sur une étagère de \(5\) places, et trois vendeurs les y prennent. queue.Queue(maxsize=5) est cette étagère : put attend une place libre, get attend un pain. Quand les boulangers ont fini, on dépose un None par vendeur : c’est le signal de fin.

Cellule 9

etagere = queue.Queue(maxsize=5)
servis = [0, 0, 0]
verrou_v = threading.Lock()

def boulanger(i):
    for k in range(5):
        etagere.put(f"pain-{i}-{k}")              # attend si l'étagère est pleine
        time.sleep(random.uniform(0.001, 0.005))

def vendeur(j):
    while (pain := etagere.get()) is not None:    # None : signal de fin
        time.sleep(random.uniform(0.002, 0.008))
        with verrou_v:
            servis[j] += 1

boulangers = [threading.Thread(target=boulanger, args=(i,)) for i in range(2)]
vendeurs = [threading.Thread(target=vendeur, args=(j,)) for j in range(3)]
for t in boulangers + vendeurs: t.start()
for t in boulangers: t.join()
for _ in vendeurs: etagere.put(None)              # un signal de fin par vendeur
for t in vendeurs: t.join()
print("pains vendus par vendeur :", servis, "  total :", sum(servis))

Résultat →

pains vendus par vendeur : [3, 3, 4]   total : 10
NoteLecture

Dix pains produits, dix vendus ; la répartition entre vendeurs change d’une exécution à l’autre. Avec un seul None pour trois vendeurs, deux vendeurs attendraient indéfiniment. Le chargeur de données de PyTorch (DataLoader) fonctionne ainsi : des processus préparent les lots d’images, la boucle d’entraînement les consomme.

AstuceÀ faire

Ralentir les vendeurs (random.uniform(0.02, 0.03)) et afficher, dans boulanger, la taille de l’étagère après chaque put (etagere.qsize()). L’étagère dépasse-t-elle \(5\) pains ?

Cellule 10 — à compléter

# refaire la cellule avec des vendeurs lents

Partie 5 — Le pipeline

Trois étapes à la chaîne, chacune dans son fil, reliées par des files : charger (\(2\) ms), filtrer (\(5\) ms), sauvegarder (\(3\) ms), pour \(100\) images. Chaque étape traite une image pendant que la suivante traite la précédente. La Séance 3 prévoit \(T = (2+5+3) + (100-1)\times5 = 505\) ms : après la mise en route, l’étape la plus lente fixe le rythme.

Cellule 11

def etape(duree, entree, sortie):
    while (x := entree.get()) is not None:      # None : signal de fin
        time.sleep(duree)                       # le travail de l'étape
        sortie.put(x)
    sortie.put(None)                            # transmettre le signal de fin

def pipeline(durees, n=100):
    q = [queue.Queue() for _ in range(len(durees) + 1)]
    etapes = [threading.Thread(target=etape, args=(d, q[k], q[k + 1]))
              for k, d in enumerate(durees)]
    t0 = time.perf_counter()
    for e in etapes: e.start()
    for x in range(n): q[0].put(x)
    q[0].put(None)
    for e in etapes: e.join()
    return time.perf_counter() - t0

d = (0.002, 0.005, 0.003)
modele = sum(d) + 99 * max(d)
print(f"mesuré : {pipeline(d):.3f} s   modèle : {modele:.3f} s")
print(f"à la suite, sans pipeline : {100 * sum(d):.3f} s")

Résultat →

mesuré : 0.528 s   modèle : 0.505 s
à la suite, sans pipeline : 1.000 s
NoteLecture

La durée mesurée est proche du modèle et environ deux fois plus courte que le traitement à la suite (\(1\) s). Elle est un peu plus longue que le modèle, car time.sleep dure toujours un peu plus que demandé.

AstuceÀ faire

Rééquilibrer les étapes à \(4\), \(3\) et \(3\) ms (la même durée totale de \(10\) ms). Prédire la durée avec le modèle, puis mesurer.

Cellule 12 — à compléter

# pipeline((0.004, 0.003, 0.003))

Partie 6 — L’interblocage

Fatou vire de l’argent à Amadou pendant qu’Amadou vire à Fatou. Chaque virement verrouille d’abord le compte débité, puis le compte crédité. Pour ne pas bloquer le notebook, la seconde attente est limitée à \(0{,}5\) s (acquire(timeout=0.5)).

Cellule 13

comptes = {"Fatou": threading.Lock(), "Amadou": threading.Lock()}

def virement(de, vers, issue):
    with comptes[de]:                                  # 1. verrouiller le compte débité
        time.sleep(0.05)                          # l'autre fil prend son premier verrou
        if comptes[vers].acquire(timeout=0.5):    # 2. le compte crédité, 0,5 s au plus
            comptes[vers].release()
            issue[de] = "fait"
        else:
            issue[de] = "bloqué, attend le compte de " + vers

issue = {}
lancer(virement, [("Fatou", "Amadou", issue), ("Amadou", "Fatou", issue)])
print(issue)

Résultat →

{'Fatou': 'bloqué, attend le compte de Amadou', 'Amadou': 'fait'}
NoteLecture

Chaque fil tient un verrou et attend celui de l’autre : un cycle d’attente, l’interblocage. Le délai fait abandonner au moins un des deux ; s’il rend son verrou assez tôt, l’autre passe. Sans délai, les deux fils attendraient indéfiniment et la cellule resterait bloquée sur {[}*{]}.

Cellule 14

def virement_ordonne(de, vers, issue):
    premier, second = sorted([de, vers])          # ordre global : alphabétique
    with comptes[premier]:
        time.sleep(0.05)
        with comptes[second]:
            issue[de] = "fait"

issue = {}
lancer(virement_ordonne, [("Fatou", "Amadou", issue), ("Amadou", "Fatou", issue)])
print(issue)

Résultat →

{'Fatou': 'fait', 'Amadou': 'fait'}
NoteLecture

Les deux virements prennent les verrous dans le même ordre (Amadou, puis Fatou) : le second fil attend le premier verrou sans rien tenir, et aucun cycle ne peut se former. Les deux virements sont faits. Règle : quand un fil prend plusieurs verrous, tous les fils les prennent dans le même ordre.

Mini-défi — Le service des relevés de notes

Un service traite \(40\) demandes de relevés de notes avec un groupe de \(10\) fils (ThreadPoolExecutor). La base de données n’accepte que \(5\) accès à la fois (lecture simulée par time.sleep(0.05)), et le nombre de demandes traitées doit être exact. Compléter la classe ; la cellule de test vérifie les deux exigences.

Cellule 15

class Service:
    def __init__(self):
        self.base = None          # à compléter : au plus 5 accès simultanés
        self.verrou = None        # à compléter : protège les compteurs
        self.traitees = 0
        self.en_cours, self.max_en_cours = 0, 0

    def traiter(self, demande):
        # à compléter : entrer dans la base, mettre à jour en_cours et max_en_cours,
        # lire les notes (time.sleep(0.05)), sortir, puis compter la demande traitée
        pass

Cellule 16

# Test : ne pas modifier
service = Service()
t0 = time.perf_counter()
with ThreadPoolExecutor(max_workers=10) as pool:
    for d in range(40):
        pool.submit(service.traiter, d)
duree = time.perf_counter() - t0
print(f"traitées : {service.traitees}   au plus en même temps : {service.max_en_cours}"
      f"   durée : {duree:.2f} s")
for nom, ok in [("40 demandes traitées", service.traitees == 40),
                ("au plus 5 accès simultanés", 1 <= service.max_en_cours <= 5)]:
    print(("respecté   : " if ok else "à corriger : ") + nom)

Résultat →

traitées : 0   au plus en même temps : 0   durée : 0.00 s
à corriger : 40 demandes traitées
à corriger : au plus 5 accès simultanés
AstuceÀ faire

Avant d’exécuter le test, prédire la durée avec la formule du sémaphore : \(40\) demandes, \(5\) places, \(0{,}05\) s par demande. Les dix fils du groupe changent-ils ce calcul ?

Ce qu’il faut retenir

  • Une suite lire-calculer-écrire peut être coupée par un autre fil : c’est une course critique.
  • Un verrou fait d’un bloc une section critique ; la vérification et l’action doivent être dans le même bloc.
  • Un sémaphore limite le nombre de fils dans une zone ; il ne protège pas les variables.
  • Une file bornée relie producteurs et consommateurs ; un signal de fin par consommateur les arrête proprement.
  • Un pipeline va au rythme de son étape la plus lente.
  • Un interblocage est un cycle d’attente ; un ordre global des verrous le rend impossible.

Pour cette séance

Le cours

Les slides

À lire en ligne

À télécharger

Retour au sommet