Dr. El Hadji Bassirou TOURÉ · DMI · FST · UCAD
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.
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")prêtcompteur = 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é.
3 RESUME 0
5 LOAD_GLOBAL 0 (compteur)
LOAD_CONST 1 (1)
BINARY_OP 0 (+)
STORE_GLOBAL 0 (compteur)
RETURN_CONST 0 (None)Lecture. LOAD_GLOBAL lit, BINARY_OP ajoute, STORE_GLOBAL écrit : trois instructions, entre lesquelles Python peut passer la main à un autre fil.
attendu : 2000 obtenu : 1001Lecture. 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.
Un verrou (threading.Lock) laisse entrer un seul fil à la fois dans le bloc with verrou:. Ce bloc est une section critique.
attendu : 2000 obtenu : 2000Lecture. Le compte est exact : quand un fil cède la main au milieu, l’autre attend devant with verrou: que le premier ait écrit.
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.
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")sans verrou : servis ['Fatou', 'Amadou'], versé 16000 F, solde 2000 F
avec verrou : servis ['Fatou'], versé 8000 F, solde 2000 FLecture. 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é.
À 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 ?
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.
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")au plus 3 voitures en même temps ; durée 0.80 sLecture. 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.
À 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.
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.
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))pains vendus par vendeur : [3, 3, 4] total : 10Lecture. 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.
À 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 ?
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.
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")mesuré : 0.528 s modèle : 0.505 s
à la suite, sans pipeline : 1.000 sLecture. 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é.
À 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.
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)).
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){'Fatou': 'bloqué, attend le compte de Amadou', 'Amadou': 'fait'}Lecture. 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 [*].
{'Fatou': 'fait', 'Amadou': 'fait'}Lecture. 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.
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.
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# 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)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ésPour finir. 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 ?