Servir plusieurs clients, interroger plusieurs machines
BGD701 — Télécom Paris
Tant que le serveur est en (2), (3) ou (4), il n’est pas en (1).
Le client suivant attend dans la file du système — le 5 de listen(5).
Exercice 1.5 : avec un time.sleep(10), le second client attend dix secondes de plus que le premier.
Pourquoi c’est important
Un site web bâti ainsi, avec 200 ms par page, sert 5 visiteurs par seconde. Le centième attend vingt secondes.
La concurrence n’est pas une optimisation. C’est une nécessité.
Un client après l’autre.
C’est ce que nous avons écrit.
Plusieurs clients à la fois.
C’est ce que nous voulons.
import threading
import time
def travail(nom, duree):
print(f"{nom} commence")
time.sleep(duree)
print(f"{nom} termine")
# Sans thread : 3 secondes au total
travail("A", 1)
travail("B", 1)
travail("C", 1)
# Avec threads : 1 seconde au total
threads = []
for nom in ["A", "B", "C"]:
t = threading.Thread(target=travail, args=(nom, 1))
t.start() # lance le thread, ne bloque pas
threads.append(t)
for t in threads:
t.join() # attend la fin de ce threadThread(target=fonction, args=(…,)) crée le threadstart() le démarre et rend la main immédiatementjoin() attend qu’il soit finiLe piège de la virgule
args=(nom) ne marche pas : ce sont de simples parenthèses, pas un tuple.
Écrivez args=(nom,) — ou args=[nom].
import socket
import threading
import os
PORT = 50007
def servir_client(connexion, adresse):
"""Execute dans un thread dedie a ce client."""
try:
charge1, charge5, charge15 = os.getloadavg()
message = f"{charge1:.2f} {charge5:.2f} {charge15:.2f}\n"
connexion.sendall(message.encode())
except OSError as e:
print(f"Erreur avec {adresse} : {e}")
finally:
connexion.close()
serveur = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serveur.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
serveur.bind(("0.0.0.0", PORT))
serveur.listen(64)
while True:
connexion, adresse = serveur.accept()
threading.Thread(target=servir_client,
args=(connexion, adresse),
daemon=True).start()
# La boucle repart immediatement vers accept()daemon=True — ce thread n’empêche pas le programme de se terminer. Sans lui, Ctrl+C laisse le processus vivant tant qu’un client est connecté.
try / finally — la connexion est fermée quoi qu’il arrive.
Oubliez de fermer, et le serveur atteint la limite de descripteurs : OSError: [Errno 24] Too many open files.
os.getloadavg() n’existe pas
Fonction Unix. Elle marche à l’école et sur macOS, pas sous Windows.
Solution de repli dans le cours, en bibliothèque standard uniquement.
Aucune bibliothèque externe : en examen le réseau est coupé, rien ne s’installe. Et testez sur une machine de l’école au moins une fois.
compteur = compteur + 1 n’est pas atomique : lire, ajouter, écrire.
Deux threads lisent 10 en même temps, tous deux écrivent 11. Un client a disparu du comptage.
C’est une condition de course (race condition).
À retenir
Toute variable modifiée par plusieurs threads doit être protégée par un verrou. Mieux encore : évitez les variables partagées.
Python a un Global Interpreter Lock : deux threads ne peuvent pas exécuter du calcul Python en même temps.
Les threads n’accélèrent donc pas les calculs. Ils excellent pour l’attente : réseau, disque, sleep.
Un serveur passe son temps à attendre : les threads sont parfaits ici. Pour paralléliser du calcul, ce sera multiprocessing, en séance 6.
À 200 ms par machine : secondes.
Pendant ces 20 secondes, le programme ne calcule rien. Il attend. Il pourrait attendre les cent machines en même temps.
import socket
from concurrent.futures import ThreadPoolExecutor
PORT = 50007
DELAI = 3.0 # timeout en secondes
def interroger(machine):
"""Renvoie (machine, charge1min) ou (machine, None) si echec."""
try:
client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
client.settimeout(DELAI)
client.connect((machine, PORT))
reponse = client.recv(1024).decode().strip()
client.close()
charge1 = float(reponse.split()[0])
return (machine, charge1)
except (OSError, ValueError, IndexError):
return (machine, None)
machines = [l.strip() for l in open("machines.txt") if l.strip()]
with ThreadPoolExecutor(max_workers=50) as pool:
resultats = list(pool.map(interroger, machines))ThreadPoolExecutor(max_workers=50) — un réservoir de 50 threads. Dès qu’un se libère, il prend la tâche suivante.
pool.map(fonction, liste) — renvoie les résultats dans l’ordre de la liste d’entrée, pas dans l’ordre d’arrivée.
with — attend la fin de toutes les tâches.
100 interrogations en 0,4 seconde au lieu de 20.
Sans elle, une machine éteinte peut bloquer connect() plusieurs minutes.
Avec elle, toute opération qui n’aboutit pas en 3 secondes lève une exception — qui hérite de OSError, donc attrapée par notre except.
Dans un système distribué
La question n’est jamais « est-ce que ça peut échouer ? »
mais « que fait mon programme quand ça échoue ? »
Un programme sans timeout n’a pas de réponse : il attend indéfiniment. Tout appel réseau doit avoir un timeout.
Les machines de TP ne sont pas joignables depuis Internet.
Il faut être sur campus-telecom ou sur le VPN
campus-telecom : d’accordeduroam : non, considéré comme un réseau extérieurC’est la cause numéro un des « ça ne marche pas chez moi ».
Suffisant pour commencer, mais rien ne garantit que ces machines existent ni qu’elles soient allumées.
https://tp.telecom-paris.fr/ — alimenté par une petite API JSON :
| Position | Signification |
|---|---|
| 0 | nom court |
| 1 | allumée ? |
| 2 | sessions locales |
| 3 | bureau à distance |
| 4 | sessions SSH |
import json
import urllib.request
URL = "https://tp.telecom-paris.fr/ajax.php"
def machines_libres():
"""Machines allumees et sans session locale, en nom complet."""
with urllib.request.urlopen(URL, timeout=10) as reponse:
donnees = json.load(reponse)
libres = []
for nom, allumee, locales, bureau, ssh in donnees["data"]:
if allumee and not locales:
libres.append(nom + ".enst.fr")
return sorted(libres)
if __name__ == "__main__":
noms = machines_libres()
with open("machines.txt", "w") as f:
f.write("\n".join(noms) + "\n")
print(f"{len(noms)} machines ecrites dans machines.txt")Une requête, une réponse JSON, un format convenu à l’avance, un timeout. Nous formaliserons tout cela en séance 3.
urllib.request et json sont dans la bibliothèque standard : rien à installer.
Une question de courtoisie
and not locales écarte les machines devant lesquelles quelqu’un est assis. Lancer trente processus sur le poste d’un camarade en train de travailler est un excellent moyen de se faire remarquer.
Ne lancez pas ssh-keygen
Une paire de clés a été configurée lors du cours précédent. La régénérer l’écrase, et vous perdez l’accès de l’autre cours.
Vérifiez, puis testez :
Si on vous demande un mot de passe : ne créez rien, venez me voir.
campus-telecom ou sur le VPNLa gestion des clés y diffère un peu, et nous ne nous y attarderons pas. Si vous butez dessus : passez sur une machine de l’école, tout y est déjà configuré.
~/.ssh/config
Attention
Ceci désactive une protection contre l’attaque de l’intercepteur. Acceptable sur le réseau fermé d’une salle de TP. Jamais en production.
ssh — exécuter une commande ailleurs
scp — copier un fichier ; la destination s’écrit machine:chemin
Le répertoire personnel étant partagé, une seule copie suffit : le fichier apparaît sur toutes les machines.
Un programme lancé par ssh est tué quand la connexion se ferme.
| Élément | Rôle |
|---|---|
nohup |
no hangup : ignorer le signal de fin de session |
& |
arrière-plan, rend la main tout de suite |
> /tmp/charge-\$USER.log 2>&1 |
sortie et erreurs dans un fichier |
/tmp est partagé par tous
Sans \$USER dans le nom, le fichier créé par un camarade vous bloque : Permission denied.
Le \ compte : sans lui, $USER serait évalué sur votre machine.
La redirection n’est pas optionnelle
Sans elle, ssh semble figé : il attend la fermeture du flux de sortie, qui n’arrivera jamais.
Arrêtez toujours vos serveurs. Un processus oublié occupe un port, et le camarade qui le choisira demain obtiendra Address already in use sans comprendre pourquoi.
sleep(5) : 5 s pour tous, pas 5, 10 et 15.ssh, ajoutez trois machines éteintes, chronométrez.vivantes.txt — il servira en séance 7.daemon=True, et finally: close()ThreadPoolExecutor + pool.map pour interroger machinescampus-telecom ou le VPN