Aide-mémoire

Usage

Cette annexe rassemble le code et les formules à avoir sous les yeux pendant les évaluations. Tout y est directement recopiable.

Sockets

Serveur	Client
`s = socket.socket(AF_INET, SOCK_STREAM)`	idem
`s.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)`	—
`s.bind(("0.0.0.0", PORT))`	—
`s.listen(64)`	—
`c, adr = s.accept()`	`s.connect((HOTE, PORT))`
`c.recv(n)` / `c.sendall(b)`	`s.recv(n)` / `s.sendall(b)`
`c.close()`	`s.close()`

import socket
serveur = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serveur.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serveur.bind(("0.0.0.0", 50007))
serveur.listen(64)
connexion, adresse = serveur.accept()

Rappels. "texte".encode() $\rightarrow$ bytes ; b"octets".decode() $\rightarrow$ str. recv renvoie b"" quand le pair a fermé. settimeout(3.0) avant connect.

Concurrence

import threading
t = threading.Thread(target=fonction, args=(a, b), daemon=True)
t.start()
t.join()

verrou = threading.Lock()
with verrou:
    compteur += 1

from concurrent.futures import ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=50) as pool:
    resultats = list(pool.map(fonction, liste))   # ordre conserve

import multiprocessing            # pour le CALCUL (contourne le GIL)
if __name__ == "__main__":        # OBLIGATOIRE
    with multiprocessing.Pool(8) as pool:
        pool.map(tache, elements)

Hexadécimal et octets

Un octet $=$ 2 chiffres hexadécimaux, de 00 à FF (0 à 255). Chiffres : 0…9 puis A $=$ 10 à F $=$ 15. Le préfixe 0x signale la base 16.

$\texttt{2A} = 2 \times 16 + 10 = 42 \qquad \texttt{FF} = 15 \times 16 + 15 = 255 \qquad \texttt{0B} = 11$

0x2A                       # 42  (litteral)
int("2A", 16)              # 42  texte hexa -> entier
hex(42)                    # '0x2a'
f"{42:02X}"                # '2A'
b"\x00\x0b".hex()          # '000b'
bytes.fromhex("000b")      # b'\x00\x0b'
int.from_bytes(b"\x00\x0b", "big")   # 11
(11).to_bytes(2, "big")              # b'\x00\x0b'

Boutisme. Gros-boutiste (big endian, ordre réseau) : l’octet de plus fort poids d’abord, 00 0B pour 11. Petit-boutiste : 0B 00. Le ! de struct impose le gros-boutiste.

Tailles des entiers dans struct. Majuscule = non signé.

Lettre	`B`	`H`	`I`	`Q`
Octets	1	2	4	8
Maximum	255	65 535	4 294 967 295	$1{,}8 \times 10^{19}$

On enchaîne les lettres : "!HB" = 2 octets puis 1 octet, soit 3 en tout. struct.calcsize("!HB") le confirme sans compter à la main.

Protocole : framing par longueur préfixée

import json, struct

TAILLE_ENTETE = 4
TAILLE_MAX = 100 * 1024 * 1024

def envoyer_message(sock, objet):
    charge = json.dumps(objet).encode("utf-8")
    sock.sendall(struct.pack("!I", len(charge)) + charge)

def recevoir_exactement(sock, nb_octets):
    morceaux = b""
    while len(morceaux) < nb_octets:
        morceau = sock.recv(nb_octets - len(morceaux))
        if not morceau:
            raise ConnectionError("Connexion fermee prematurement")
        morceaux += morceau
    return morceaux

def recevoir_message(sock):
    entete = recevoir_exactement(sock, TAILLE_ENTETE)
    (taille,) = struct.unpack("!I", entete)
    if taille > TAILLE_MAX:
        raise ValueError(f"Message trop grand : {taille}")
    charge = recevoir_exactement(sock, taille)
    return json.loads(charge.decode("utf-8"))

Squelette de serveur concurrent

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

def servir(connexion, adresse):
    connexion.settimeout(30.0)
    try:
        while True:
            requete = recevoir_message(connexion)
            if requete.get("type") == "QUIT":
                break
            envoyer_message(connexion, traiter(requete))
    except (ConnectionError, OSError, ValueError, KeyError):
        pass
    finally:
        connexion.close()

while True:
    c, a = serveur.accept()
    threading.Thread(target=servir, args=(c, a),
                     daemon=True).start()

Partitionnement et MapReduce

import hashlib

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

def mapreduce(fichiers, f_map, f_reduce):
    intermediaires = []
    for nom in fichiers:                                    # MAP
        contenu = open(nom, encoding="utf-8",
                       errors="ignore").read()
        intermediaires.extend(f_map(nom, contenu))
    groupes = {}                                            # SHUFFLE
    for cle, valeur in intermediaires:
        groupes.setdefault(cle, []).append(valeur)
    return {c: f_reduce(c, v) for c, v in groupes.items()}   # REDUCE

Écriture atomique.

with open(chemin + ".tmp", "w", encoding="utf-8") as f:
    json.dump(donnees, f)
os.replace(chemin + ".tmp", chemin)

Dossier local propre.

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

Faire évoluer un protocole.

valeur = requete.get("champ", valeur_par_defaut)   # jamais requete[...]
# 1. tout nouveau champ est optionnel et a une valeur par defaut
# 2. on ne renomme ni ne supprime jamais un champ obligatoire
# 3. un champ inconnu s'ignore, il ne se rejette pas

L’ancêtre Unix de MapReduce.

$ cat *.txt | tr -c '[:alpha:]' '\n' | tr A-Z a-z \
      | sort | uniq -c | sort -rn | head -20
#   map                         shuffle   reduce

Mesure et formules

import time
t0 = time.perf_counter()
...
duree = time.perf_counter() - t0

def percentile(valeurs, p):
    t = sorted(valeurs)
    return t[int(round((p / 100.0) * (len(t) - 1)))]

Grandeur	Formule
Accélération	$S(p) = \dfrac{T(1)}{T(p)}$
Amdahl, forme générale ( $a$ : part améliorée, $k$ : facteur)	$S = \dfrac{1}{(1-a) + \dfrac{a}{k}}$
Loi d’Amdahl	$S(p) = \dfrac{1}{f + \dfrac{1-f}{p}}$
Plafond	$S(\infty) = \dfrac{1}{f}$
Estimer $f$	$f = \dfrac{1/S(p) - 1/p}{1 - 1/p}$

Attention

Une courbe d’accélération n’a de sens qu’à jeu de données constant à tous les points de mesure.

Déploiement

# UN seul scp (le home est partage par NFS)
$ scp travailleur.py protocole.py login@tp-1a201-01.enst.fr:~/bgd701/

# Votre port personnel : 40000 + uid % 20000
$ id -u

# Lancer, UNE machine a la fois (deployer.py le fait pour vous)
$ ssh MACHINE "cd ~/bgd701 && nohup python3 travailleur.py \
      > /tmp/bgd701-travailleur-\$USER.log 2>&1 &"

# Nettoyage : -u (vos processus), [t] (ne pas tuer son propre shell),
# apostrophes (sinon zsh : "no matches found")
$ ssh MACHINE "pkill -u \$USER -f '[t]ravailleur.py' ; rm -rf /tmp/bgd701-\$USER"

# Avec le script de la seance 7
$ python3 deployer.py deployer 4      # ecrit travailleurs.txt
$ python3 deployer.py nettoyer

# Inspecter les disques d'une machine
$ df -h          # espaces disponibles
$ mount          # ceux qui contiennent ':' viennent du reseau

Erreurs fréquentes et diagnostic

Message	Cause et remède
`Address already in use`	Port occupé. Ajouter `SO_REUSEADDR`, ou changer de port, ou tuer l’ancien processus. Sur une machine partagée, un camarade l’utilise peut-être : prendre un port personnel, `40000 + uid % 20000`.
`Connection refused`	Personne n’écoute : serveur non lancé, mauvais port, ou serveur sur `127.0.0.1` au lieu de `0.0.0.0`.
`a bytes-like object is required`	`.encode()` oublié avant `sendall`.
`ModuleNotFoundError: No module named 'protocole'`	Le fichier n’est pas dans le même dossier que le programme lancé, ou il ne s’appelle pas exactement `protocole.py`.
`Too many open files`	Connexions non fermées : ajouter `finally: connexion.close()`.
Le programme reste figé	`settimeout` manquant, ou `recv` qui attend des octets qui ne viendront pas.
Boucle à 100 % de CPU	`recv` renvoie `b""` et le cas n’est pas testé.
Comptages faux une fois distribué	`hash()` natif au lieu de `hashlib`.
Le résultat change à chaque exécution	Variable partagée entre threads sans verrou.
La machine se fige au lancement	`if __name__ == "__main__"` manquant avec `multiprocessing`.
Phase map très lente	Fichiers intermédiaires écrits sur le NFS au lieu de `/tmp`.
`zsh: no matches found: [t]ravailleur.py`	Motif entre crochets sans apostrophes : zsh le prend pour des noms de fichiers et n’exécute rien. Écrire `'[t]ravailleur.py'`.
`ssh` renvoie 255 après un `pkill -f`	`pkill` a tué le shell distant, dont la ligne de commande contient le motif : la suite n’est pas exécutée. Utiliser `[t]ravailleur.py`.

Les réflexes à avoir

À retenir

Un settimeout sur tout appel réseau.
Un try/except autour de tout appel réseau.
Un finally: close() pour toute connexion.
Jamais recv seul : toujours recevoir_exactement.
Jamais hash() : toujours hashlib.
Jamais le NFS pour les intermédiaires : toujours /tmp.
Jamais la moyenne seule : toujours les percentiles.
Jamais time.time() pour chronométrer : toujours perf_counter().
Jamais une courbe d’accélération à jeu de données variable.
Le script de nettoyage s’écrit avant d’en avoir besoin.
Le résultat distribué se vérifie contre un calcul mono-machine.