8 Évaluation 2

Déroulement de la séance

Évaluation sur machine : 3 heures, soit la totalité de la séance.
Le programme porte sur l’ensemble du cours, séances 1 à 7.

8.1 Modalités

Identiques à celles de l’évaluation 1 (séance 4) : 3 heures, sur machine, réseau coupé, ce cours et vos notes manuscrites autorisés, ni Internet ni assistant IA. Barème : 10 points pour les questions, 10 points pour la programmation.

Le rendu suit exactement le même format : un unique fichier texte BGD701_EVAL2_NOM_Prenom.txt contenant les réponses et tout le code, sous les balises décrites en séance 4. Seule la liste des balises FICHIER change ; elle est donnée dans le sujet.

Ce sur quoi vous serez évalué

L’évaluation 2 suppose acquis les réflexes de l’évaluation 1 (framing, concurrence, timeouts) : ils font partie des critères de la partie programmation. Les questions portent surtout sur les séances 5 à 7 : partitionnement, réplication, MapReduce, mesure et loi d’Amdahl.

8.2 Examen blanc n^o2

Ce que cet examen blanc est, et ce qu’il n’est pas

Comme pour le premier, ce sujet reproduit la forme de l’évaluation, pas son contenu. Le jour de l’évaluation, l’application à construire et les questions seront différentes. Ce qui se transpose, c’est l’infrastructure (framing, serveur concurrent, maître qui distribue et relance) et la façon de raisonner sur le partitionnement, les pannes et la mesure.

Le corrigé est en annexe, Section 8.3.

Partie 1 — Questions (10 points)

Question 1 — Mesurer honnêtement 1,5 pt

Un service traite 10 000 requêtes. 9 900 prennent 20 ms, 90 prennent 500 ms, et 10 prennent 8 s.

Calculez la latence moyenne.
Donnez le p50 et le p99 (une justification suffit, pas besoin de calcul exact).
Le responsable technique affirme : « notre latence moyenne est excellente, il n’y a pas de problème ». Que lui répondez-vous, en deux phrases ?
Une page appelle 10 services de ce type en parallèle et attend les 10 réponses. Quelle est approximativement la probabilité que la page soit servie en 500 ms ou plus ?

Question 2 — Répliquer et partitionner 1,5 pt

En une phrase chacune : quelle est la différence entre réplication et partitionnement ? Quel problème chacune résout-elle ?
Un système stocke 40 To de données sur des machines munies de 4 To de disque, avec un facteur de réplication de 3. Combien de machines faut-il au minimum ?
Un utilisateur publie un message puis ne le voit pas en rechargeant la page. Quel mécanisme est en cause ? Donnez son nom et une parade.

Question 3 — Partitionnement par hachage 1,5 pt

def partition(cle, N):
    return hash(cle) % N

Ce code contient un bug grave qui ne se manifeste que lorsque le système est distribué sur plusieurs machines. Lequel ? Corrigez-le.
On partitionne 10 000 clés sur $N = 10$ machines, puis on ajoute une machine ( $N = 11$ ). Quel ordre de grandeur du pourcentage de clés change de partition ? Justifiez brièvement.
Les données sont partitionnées par identifiant de vendeur sur une place de marché. Une machine sur vingt est à 100 % de CPU, les autres à 5 %. Comment s’appelle ce phénomène ? Le hachage suffit-il à l’éviter ? Justifiez.
Citez un avantage du partitionnement par plage sur le partitionnement par hachage.

Question 4 — MapReduce 1,5 pt

Donnez les fonctions map et reduce (en pseudo-code ou en Python) qui, à partir de fichiers de log de la forme "2026-03-14 ERREUR message", comptent le nombre d’événements par niveau (INFO, ERREUR…).
Avec $M = 200$ tâches map et $R = 10$ tâches reduce, combien de fichiers intermédiaires sont produits ?
Qu’est-ce qu’un combiner ? Donnez une fonction reduce pour laquelle il est valide, et une pour laquelle il ne l’est pas, en justifiant.
Pourquoi aucun réducteur ne peut-il démarrer avant la fin de tous les mappeurs ? Quel est le nom de ce point de synchronisation, et quelle conséquence a-t-il sur la performance ?

Question 5 — NFS et disque local 1 pt

Un groupe déploie 100 travailleurs qui écrivent leurs fichiers intermédiaires dans ~/tmp_mapreduce/. Le calcul est très lent et les autres utilisateurs de la salle se plaignent.

Expliquez précisément ce qui se passe.
Quelle correction apportez-vous ? Une seule ligne suffit.
Après correction, une machine tombe en panne alors que sa tâche map était terminée. Faut-il la refaire ? Pourquoi ?
Même question pour un reduce terminé.

Question 6 — Loi d’Amdahl 1 pt

Un calcul dure 400 s sur une machine et 50 s sur 16 machines.

Calculez l’accélération $S(16)$ .
Estimez la fraction séquentielle $f$ (formule : $f = (1/S - 1/p)/(1 - 1/p)$ ).
Quel est le plafond d’accélération, même avec un nombre infini de machines ?
Le groupe veut passer à 200 machines. Est-ce une bonne idée ? Que conseillez-vous à la place ?

Question 7 — Critique d’une expérience 1 pt

Un binôme présente ce tableau pour justifier son accélération :

Machines	Fichiers traités	Temps
1	10	120 s
10	100	150 s
50	500	160 s

Ils concluent : « notre système passe très bien à l’échelle ».

Cette conclusion est-elle justifiée par ces mesures ? Pourquoi ?
Que mesure réellement ce tableau ?
Décrivez le protocole expérimental correct pour tracer une courbe d’accélération.

Question 8 — Détecter une panne 1 pt

Peut-on distinguer avec certitude une machine en panne d’une machine très lente ? Justifiez en une phrase.
Un délai de détection très court : quel risque ? Un délai très long : quel risque ?
Comment s’appelle la situation où l’on croit un leader mort, on en élit un nouveau, et l’ancien revient ?

Partie 2 — Programmation (10 points)

Un MapReduce distribué réduit

On veut compter les mots d’un ensemble de fichiers avec un maître et plusieurs travailleurs sur localhost, sur des ports différents.

1. travailleur.py 5 pts

Un serveur concurrent, utilisant le framing par longueur préfixée, qui répond à :

Requête	Effet
`{"type":"PING"}`	`{"type":"PONG"}`
`{"type":"MAP","id_map":i,"fichier":f,"R":R}`	Lit le fichier, compte les mots, répartit par `hashlib.md5(mot) % R`, écrit $R$ fichiers dans `/tmp/eval-$USER/`, répond `OK`
`{"type":"PARTITION","id_map":i,"r":r}`	Renvoie le contenu du fichier correspondant

2. maitre.py 5 pts

Un programme qui :

lit une liste de travailleurs (127.0.0.1:port) et une liste de fichiers ;
vérifie par PING lesquels répondent, et n’utilise que ceux-là ;
distribue les tâches MAP en parallèle, avec un timeout ;
attend la fin de tous les maps (barrière) ;
pour chaque $r$ de 0 à $R-1$ , récupère les partitions $r$ chez tous les mappeurs, les fusionne, et applique le reduce côté maître (pour simplifier) ;
affiche les 10 mots les plus fréquents et le temps de chaque phase.

Robustesse (compte dans la note). Si un travailleur ne répond pas pendant la phase map, sa tâche doit être relancée sur un autre travailleur. Testez-le : lancez le calcul, tuez un travailleur en cours de route, et vérifiez que le résultat final reste correct.

Vérification. Votre résultat doit être identique à celui d’un simple comptage mono-processus sur les mêmes fichiers. Fournissez le petit script de vérification.

8.3 Corrigés

Partie 1

Question 1 — Mesurer honnêtement

$(9900 \times 20 + 90 \times 500 + 10 \times 8000)/10000 = 32{,}3$ ms.
Le p50 vaut 20 ms (la grande majorité des requêtes). Le p99 est la 100^e valeur la plus lente : les 10 requêtes à 8 s et 90 à 500 ms représentent exactement le dernier centile, donc p99 $\approx$ 500 ms.
La moyenne de 32 ms cache le fait qu’une requête sur cent dépasse une demi-seconde et qu’une sur mille prend huit secondes. Ce sont ces requêtes-là que les utilisateurs remarquent et signalent.
La page est lente si au moins un des 10 appels l’est. La probabilité qu’aucun ne dépasse 500 ms est $0{,}99^{10} \approx 0{,}90$ , donc environ 10 % des pages sont lentes — alors que chaque service pris isolément n’est « lent » que dans 1 % des cas.

Question 2 — Répliquer et partitionner

La réplication met la même donnée sur plusieurs machines : elle résout la disponibilité et la montée en charge des lectures. Le partitionnement met des données différentes sur des machines différentes : il résout le volume et la montée en charge des écritures.
$40 \times 3 = 120$ To à stocker, sur des disques de 4 To : il faut au minimum 30 machines (en pratique davantage, pour la marge et le rééquilibrage).
Le retard de réplication (replication lag) : la lecture a été servie par un suiveur pas encore à jour. Parade : read-your-writes, c’est-à-dire lire sur le leader pendant quelques secondes après une écriture.

Question 3 — Partitionnement par hachage

hash() est salée aléatoirement à chaque démarrage de l’interpréteur : deux machines calculent des partitions différentes pour la même clé. Correction :

int(hashlib.md5(cle.encode()).hexdigest(), 16) % N

Environ 90 % des clés changent de partition (plus précisément $1 - 1/11 \approx 91\,\%$ ) : les restes modulo 10 et modulo 11 ne coïncident que dans environ un cas sur onze.
C’est un point chaud (hot spot). Le hachage n’y remédie pas : il répartit uniformément les clés, mais toute la charge de ce vendeur porte sur une seule clé, qui va nécessairement sur une seule machine.
Le partitionnement par plage permet les requêtes par intervalle efficaces (par exemple « toutes les commandes entre le 1^er et le 15 mars »), qui obligeraient sinon à interroger toutes les partitions.

Question 4 — MapReduce

def fonction_map(nom, contenu):
    for ligne in contenu.splitlines():
        morceaux = ligne.split()
        if len(morceaux) >= 2:
            yield (morceaux[1], 1)      # "2026-03-14 ERREUR message"
                                        #       0       1       2
def fonction_reduce(cle, valeurs):
    return sum(valeurs)

$200 \times 10 =$ 2000 fichiers intermédiaires.
Un combiner est une agrégation locale appliquée à la sortie de map avant écriture, afin de réduire le volume du shuffle. Il est valide pour la somme (associative et commutative), invalide pour la moyenne : la moyenne des moyennes n’est pas la moyenne.
Parce qu’un réducteur doit disposer de toutes les valeurs d’une clé, et qu’une tâche map encore en cours pourrait en produire d’autres. Ce point s’appelle la barrière. Conséquence : le système avance à la vitesse du mappeur le plus lent, et cette attente constitue une part irréductiblement séquentielle du calcul (le $f$ de la loi d’Amdahl).

Question 5 — NFS et disque local

Le répertoire personnel est monté par NFS : les 100 machines écrivent en réalité sur un seul disque, celui du serveur NFS, à travers le réseau. Le parallélisme des écritures est annulé, la latence réseau s’ajoute, et le serveur partagé sature pour tous les utilisateurs.
Écrire dans /tmp : par exemple DOSSIER = f"/tmp/mr-{getpass.getuser()}".
Oui, il faut le refaire. Ses fichiers intermédiaires étaient sur le disque local de la machine morte, donc inaccessibles.
Non. Le résultat d’un reduce terminé a déjà été collecté par le maître (ou écrit dans un espace partagé), il n’est plus sur la machine perdue.

Question 6 — Loi d’Amdahl

$S(16) = 400/50 =$ 8.
$f = (1/8 - 1/16)/(1 - 1/16) = (0{,}125 - 0{,}0625)/0{,}9375 =$ 0,0667, soit environ 6,7 %.
Plafond $= 1/f \approx$ 15.
Non : passer de 16 à 200 machines ferait passer l’accélération d’environ 8 à environ 14, pour douze fois plus de ressources — et en pratique la courbe redescendrait. Il vaut mieux réduire $f$ : supprimer une synchronisation, découper les entrées plus finement, recouvrir le shuffle avec le calcul.

Question 7 — Critique d’une expérience

Non. Le jeu de données change à chaque ligne : on ne peut pas comparer les temps.
Ce tableau mesure la scalabilité faible (weak scaling) : la capacité à traiter un volume croissant en gardant un temps à peu près constant. C’est une propriété intéressante — et d’ailleurs plutôt bonne ici — mais ce n’est pas une accélération.
Protocole correct : fixer un jeu de données unique, mesurer $T(p)$ pour $p = 1, 2, 4, 8, \ldots$ sur ce même jeu, répéter chaque point au moins trois fois et garder la médiane, puis tracer $S(p) = T(1)/T(p)$ .

Question 8 — Détecter une panne

Non. Un silence peut venir d’une machine morte, d’une machine saturée ou d’un réseau coupé : de l’extérieur, ces situations sont indiscernables. C’est un résultat théorique, pas une limite technique.
Délai court : faux positifs — on déclare morte une machine vivante, on refait du travail inutilement et on risque le split brain. Délai long : on garde longtemps une machine réellement morte, les requêtes se perdent et le système reste bloqué.
Le split brain : deux leaders coexistent et acceptent des écritures contradictoires.

Partie 2

Évaluation 2 — travailleur.py

travailleur.py

import socket, threading, os, json, sys, getpass, hashlib
from protocole import envoyer_message, recevoir_message

DOSSIER = f"/tmp/eval-{getpass.getuser()}"
os.makedirs(DOSSIER, exist_ok=True)

def partition(cle, R):
    e = hashlib.md5(str(cle).encode("utf-8")).hexdigest()
    return int(e, 16) % R

def faire_map(requete):
    id_map, R = requete["id_map"], requete["R"]
    with open(requete["fichier"], encoding="utf-8",
              errors="ignore") as f:
        contenu = f.read()

    seaux = [dict() for _ in range(R)]
    for mot in contenu.split():
        mot = mot.strip(".,;:!?\"'()[]").lower()
        if not mot:
            continue
        r = partition(mot, R)
        seaux[r][mot] = seaux[r].get(mot, 0) + 1

    for r in range(R):
        chemin = f"{DOSSIER}/map-{id_map:04d}-part-{r:02d}"
        with open(chemin + ".tmp", "w", encoding="utf-8") as f:
            json.dump(seaux[r], f)
        os.replace(chemin + ".tmp", chemin)
    return {"type": "OK", "id_map": id_map}

def lire_partition(requete):
    chemin = (f"{DOSSIER}/map-{requete['id_map']:04d}"
              f"-part-{requete['r']:02d}")
    if not os.path.exists(chemin):
        return {"type": "OK", "donnees": {}}
    with open(chemin, encoding="utf-8") as f:
        return {"type": "OK", "donnees": json.load(f)}

def traiter(requete):
    t = requete.get("type")
    if t == "PING":       return {"type": "PONG"}
    if t == "MAP":        return faire_map(requete)
    if t == "PARTITION":  return lire_partition(requete)
    return {"type": "ERREUR", "message": "type inconnu"}

def servir(connexion, adresse):
    connexion.settimeout(120.0)
    try:
        while True:
            envoyer_message(connexion,
                            traiter(recevoir_message(connexion)))
    except (ConnectionError, OSError, ValueError, KeyError):
        pass
    finally:
        connexion.close()

if __name__ == "__main__":
    port = int(sys.argv[1])
    srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    srv.bind(("0.0.0.0", port))
    srv.listen(64)
    print(f"Travailleur pret sur le port {port}")
    while True:
        c, a = srv.accept()
        threading.Thread(target=servir, args=(c, a),
                         daemon=True).start()

Évaluation 2 — maitre.py

maitre.py

import socket, time
from concurrent.futures import ThreadPoolExecutor
from protocole import envoyer_message, recevoir_message

R = 4

def demander(adresse, requete, delai=30.0):
    machine, port = adresse.split(":")
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.settimeout(delai)
    s.connect((machine, int(port)))
    envoyer_message(s, requete)
    reponse = recevoir_message(s)
    s.close()
    return reponse

def est_vivant(adresse):
    try:
        return demander(adresse, {"type": "PING"},
                        5.0).get("type") == "PONG"
    except (OSError, ValueError, ConnectionError):
        return False

def phase_map(travailleurs, fichiers, essais_max=3):
    """Renvoie [(travailleur, id_map)] ; relance en cas d'echec."""
    a_faire = list(enumerate(fichiers))
    reussies = []
    vivants = list(travailleurs)

    for essai in range(essais_max):
        if not a_faire or not vivants:
            break

        def executer(tache):
            id_map, fichier = tache
            cible = vivants[id_map % len(vivants)]
            try:
                r = demander(cible, {"type": "MAP", "id_map": id_map,
                                     "fichier": fichier, "R": R}, 120.0)
                if r.get("type") == "OK":
                    return ("ok", cible, id_map, fichier)
            except (OSError, ValueError, ConnectionError):
                pass
            return ("echec", cible, id_map, fichier)

        with ThreadPoolExecutor(max_workers=len(vivants)) as pool:
            resultats = list(pool.map(executer, a_faire))

        morts = set()
        reste = []
        for etat, cible, id_map, fichier in resultats:
            if etat == "ok":
                reussies.append((cible, id_map))
            else:
                morts.add(cible)
                reste.append((id_map, fichier))

        vivants = [t for t in vivants if t not in morts]
        a_faire = reste
        if a_faire:
            print(f"Essai {essai+1} : {len(a_faire)} taches "
                  f"a relancer, {len(vivants)} travailleurs restants")

    return reussies, a_faire

if __name__ == "__main__":
    travailleurs = [l.strip() for l in open("travailleurs.txt")
                    if l.strip()]
    fichiers = [l.strip() for l in open("fichiers.txt") if l.strip()]

    t0 = time.perf_counter()
    vivants = [t for t in travailleurs if est_vivant(t)]
    print(f"{len(vivants)}/{len(travailleurs)} travailleurs vivants")

    t1 = time.perf_counter()
    sources, echecs = phase_map(vivants, fichiers)
    if echecs:
        print(f"ATTENTION : {len(echecs)} taches non terminees")
    t2 = time.perf_counter()

    # ---- BARRIERE franchie : shuffle + reduce cote maitre ----
    def lire_un_map(cible, id_map):
        """Lit les R partitions d'un map. Leve si l'une manque."""
        partiel = {}
        for r in range(R):
            donnees = demander(cible, {"type": "PARTITION",
                                       "id_map": id_map,
                                       "r": r}, 60.0)["donnees"]
            for mot, n in donnees.items():
                partiel[mot] = partiel.get(mot, 0) + n
        return partiel

    # On n'agrege un map QUE s'il a ete lu en entier : sinon on le refait.
    total, perdus = {}, []
    for cible, id_map in sources:
        try:
            partiel = lire_un_map(cible, id_map)
        except (OSError, ValueError, ConnectionError):
            perdus.append(id_map)
            continue
        for mot, n in partiel.items():
            total[mot] = total.get(mot, 0) + n

    # Un travailleur mort APRES ses maps emporte ses partitions avec lui.
    # Les ignorer donnerait un resultat faux sans aucun message d'erreur.
    if perdus:
        print(f"{len(perdus)} maps perdus : on les refait")
        vivants = [t for t in travailleurs if est_vivant(t)]
        if not vivants:
            raise RuntimeError("plus aucun travailleur : resultat impossible")
        for k, id_map in enumerate(perdus):
            cible = vivants[k % len(vivants)]
            neuf = len(fichiers) + k     # id neuf : pas de collision
            demander(cible, {"type": "MAP", "id_map": neuf,
                             "fichier": fichiers[id_map], "R": R}, 300.0)
            for mot, n in lire_un_map(cible, neuf).items():
                total[mot] = total.get(mot, 0) + n
    t3 = time.perf_counter()

    for mot, n in sorted(total.items(), key=lambda x: -x[1])[:10]:
        print(f"{n:8d}  {mot}")
    print(f"\nping   : {t1-t0:6.2f} s")
    print(f"map    : {t2-t1:6.2f} s")
    print(f"reduce : {t3-t2:6.2f} s")
    print(f"TOTAL  : {t3-t0:6.2f} s")

Script de vérification.

reference = {}
for nom in open("fichiers.txt").read().split():
    for mot in open(nom, encoding="utf-8",
                    errors="ignore").read().split():
        mot = mot.strip(".,;:!?\"'()[]").lower()
        if mot:
            reference[mot] = reference.get(mot, 0) + 1

assert reference == total, "le resultat distribue est faux"
print(f"OK : {len(reference)} cles identiques")