Threads &
Synchronisation

# I/O système (réseau, fichiers, stdin)
response = requests.get("https://...")  # ← GIL libéré
data = file.read()                        # ← GIL libéré
time.sleep(1)                            # ← GIL libéré

# Extensions C qui le libèrent explicitement
numpy_result = np.dot(a, b)  # NumPy libère le GIL

# Calcul pur Python → GIL NON libéré
total = sum(range(10_000_000))  # bloque les autres threads

import threading
import time

def telecharger(url: str, duree: float) -> None:
    """Simule un téléchargement."""
    print(f"[{threading.current_thread().name}] Début {url}")
    time.sleep(duree)            # GIL libéré ici
    print(f"[{threading.current_thread().name}] Fin {url}")

# Créer les threads
t1 = threading.Thread(
    target=telecharger,
    args=("https://site-a.com", 2),
    name="T-SiteA"
)
t2 = threading.Thread(
    target=telecharger,
    args=("https://site-b.com", 3),
    name="T-SiteB"
)

# Démarrer (non bloquant)
t1.start()
t2.start()

# Attendre la fin des deux
t1.join()
t2.join()

print("Tous les téléchargements terminés")
# Durée totale ≈ 3s (pas 5s) — parallélisme I/O

# Dans n'importe quelle fonction
t = threading.current_thread()
print(t.name)      # "T-SiteA"
print(t.ident)     # ID système
print(t.is_alive()) # True / False

# Lister tous les threads actifs
for t in threading.enumerate():
    print(t.name, t.is_alive())

import threading
import time
import requests

class WorkerThread(threading.Thread):
    def __init__(self, url: str) -> None:
        # Toujours appeler super().__init__()
        super().__init__(name=f"Worker-{url[:20]}")
        self.url     = url
        self.resultat = None
        self.erreur   = None

    def run(self) -> None:
        """Appelée par start() — NE PAS appeler directement."""
        try:
            resp = requests.get(self.url, timeout=10)
            self.resultat = resp.json()
        except Exception as e:
            self.erreur = e

# Utilisation
urls = [
    "https://api.example.com/users",
    "https://api.example.com/products",
]
workers = [WorkerThread(url) for url in urls]

for w in workers: w.start()
for w in workers: w.join()

for w in workers:
    if w.erreur:
        print(f"Erreur {w.url}: {w.erreur}")
    else:
        print(f"OK {w.url}: {w.resultat}")

class MoniteurThread(threading.Thread):
    def __init__(self) -> None:
        super().__init__(daemon=True)
        self._stop_event = threading.Event()

    def stop(self) -> None:
        self._stop_event.set()

    def run(self) -> None:
        while not self._stop_event.is_set():
            # Travail périodique
            self._stop_event.wait(5)  # sleep interruptible

import threading, time

def tache_longue():
    time.sleep(10)
    print("Fin de la tâche longue")

# Thread NON-daemon (défaut)
# → programme attend 10s même si main() est terminé
t1 = threading.Thread(target=tache_longue)

# Thread DAEMON
# → tué immédiatement à la fin du programme
t2 = threading.Thread(target=tache_longue, daemon=True)

# Ou après création :
t2.daemon = True  # avant start() uniquement

t2.start()
print("Main terminé — le daemon sera tué")
# Programme se ferme ici sans attendre t2

import threading

compteur = 0

def incrementer(n: int) -> None:
    global compteur
    for _ in range(n):
        compteur += 1  # ← PAS atomique !

t1 = threading.Thread(target=incrementer, args=(100_000,))
t2 = threading.Thread(target=incrementer, args=(100_000,))
t1.start(); t2.start()
t1.join();  t2.join()

print(compteur)
# Attendu : 200 000
# Obtenu   : 173 842 (valeur variable !)

# ✗ Lecture-modification-écriture
x += 1
liste.append(element)  # OK en Python (GIL)

# ✗ Test puis action (check-then-act)
if cle not in dico:    # Thread 2 peut s'insérer ici
    dico[cle] = valeur  # entre le test et l'écriture

# ✗ Séquence de plusieurs opérations
total = lire_solde()    # Thread 2 peut modifier
nouveau = total - 100  # entre ces deux lignes
ecrire_solde(nouveau)

import threading

class CompteurSafe:
    def __init__(self) -> None:
        self._valeur = 0
        self._lock   = threading.Lock()

    def incrementer(self) -> None:
        with self._lock:         # acquire() + release() auto
            self._valeur += 1   # section critique

    def valeur(self) -> int:
        with self._lock:
            return self._valeur  # lecture protégée aussi

# Test
compteur = CompteurSafe()
threads = [
    threading.Thread(target=lambda: [compteur.incrementer()
                                       for _ in range(100_000)])
    for _ in range(2)
]
for t in threads: t.start()
for t in threads: t.join()

print(compteur.valeur())  # Toujours 200 000 ✓

# Lock ordinaire → deadlock si même thread acquiert 2x
lock = threading.Lock()
lock.acquire()
lock.acquire()  # ← DEADLOCK ! attend éternellement

# RLock → même thread peut l'acquérir N fois
rlock = threading.RLock()

def methode_a(self):
    with self._rlock:
        self.methode_b()  # OK avec RLock

def methode_b(self):
    with self._rlock:   # même thread → pas de blocage
        ...

lock = threading.Lock()

# Tentative d'acquisition avec timeout
acquis = lock.acquire(timeout=2.0)
if acquis:
    try:
        # section critique
        ...
    finally:
        lock.release()
else:
    print("Impossible d'acquérir le lock")

import threading, time

# Autoriser au plus 3 accès simultanés à la BDD
sem_db = threading.Semaphore(3)

def requete_db(worker_id: int) -> None:
    print(f"Worker {worker_id} : en attente...")
    with sem_db:  # attend si 3 threads dedans
        print(f"Worker {worker_id} : connexion BDD")
        time.sleep(1)  # simule requête
        print(f"Worker {worker_id} : terminé")

# 10 workers → max 3 en BDD à la fois
threads = [
    threading.Thread(target=requete_db, args=(i,))
    for i in range(10)
]
for t in threads: t.start()
for t in threads: t.join()

import threading, time

class RateLimiter:
    """Limite à N appels par seconde."""
    def __init__(self, max_par_sec: int):
        self.sem = threading.BoundedSemaphore(max_par_sec)
        self.max = max_par_sec

    def appel(self, fn, *args):
        self.sem.acquire()
        try:
            return fn(*args)
        finally:
            threading.Timer(1.0, self.sem.release).start()

pret = threading.Event()

def chargement():
    time.sleep(2)         # charge les données
    pret.set()             # signal "c'est prêt"

def traitement():
    print("En attente des données...")
    pret.wait()            # bloque jusqu'à set()
    print("Traitement en cours")

threading.Thread(target=chargement).start()
threading.Thread(target=traitement).start()

# event.clear()  → remet à False
# event.is_set() → vérifie sans bloquer
# event.wait(timeout=5) → attente limitée

cond = threading.Condition()
donnees = []

def producteur():
    for i in range(5):
        with cond:
            donnees.append(i)
            cond.notify()  # réveille 1 consommateur
        time.sleep(0.5)

def consommateur():
    while True:
        with cond:
            # wait() libère le lock et attend notify()
            cond.wait_for(lambda: len(donnees) > 0)
            item = donnees.pop(0)
        print(f"Consommé : {item}")

import threading
import queue
import time

# Queue bornée : bloque le prod si pleine
q: queue.Queue = queue.Queue(maxsize=10)
POISON_PILL = None  # signal d'arrêt

def producteur(items: list) -> None:
    for item in items:
        q.put(item)          # bloque si queue pleine
        print(f"Produit : {item}")
    q.put(POISON_PILL)       # signal de fin

def consommateur() -> None:
    while True:
        item = q.get()       # bloque si queue vide
        if item is POISON_PILL:
            q.task_done()
            break
        time.sleep(0.1)    # traitement
        print(f"Consommé : {item}")
        q.task_done()        # obligatoire après get()

prod = threading.Thread(
    target=producteur, args=(["a", "b", "c", "d"],)
)
cons = threading.Thread(target=consommateur)

prod.start(); cons.start()
prod.join();  cons.join()

q.join()  # attend que task_done() soit appelé pour chaque item

from concurrent.futures import ThreadPoolExecutor, as_completed
import requests

urls = [
    "https://api.github.com/users/python",
    "https://api.github.com/users/django",
    "https://api.github.com/users/flask",
]

def fetch(url: str) -> dict:
    resp = requests.get(url, timeout=10)
    resp.raise_for_status()
    return resp.json()

── map() : simple, résultats dans l'ordre ──
with ThreadPoolExecutor(max_workers=5) as pool:
    resultats = list(pool.map(fetch, urls))
    # résultats[i] correspond à urls[i]

── submit() : plus de contrôle, gestion erreurs ──
with ThreadPoolExecutor(max_workers=5) as pool:
    futures = {pool.submit(fetch, url): url for url in urls}

    for future in as_completed(futures):
        url = futures[future]
        try:
            data = future.result()
            print(f"✓ {url}: {data['login']}")
        except Exception as e:
            print(f"✗ {url}: {e}")

import os

# I/O-bound : plus de workers que de CPUs
# (règle empirique : 4 × nb_CPUs)
nb_io = (os.cpu_count() or 1) * 4

# CPU-bound : jamais > nb de CPUs
# (préférer multiprocessing dans ce cas)
nb_cpu = os.cpu_count() or 1

# Défaut Python 3.8+ : min(32, cpu_count + 4)

from concurrent.futures import ProcessPoolExecutor
import math

def est_premier(n: int) -> bool:
    """Calcul CPU intensif."""
    if n < 2: return False
    for i in range(2, int(math.sqrt(n)) + 1):
        if n % i == 0: return False
    return True

nombres = range(10_000_000, 10_000_100)

# ThreadPoolExecutor → séquentiel (GIL)
# ProcessPoolExecutor → vrai parallélisme
with ProcessPoolExecutor() as pool:
    resultats = list(pool.map(est_premier, nombres, chunksize=10))

premiers = [n for n, p in zip(nombres, resultats) if p]
print(premiers)

import asyncio
import aiohttp

async def fetch(session, url: str) -> dict:
    # await cède le contrôle pendant l'I/O
    async with session.get(url) as resp:
        return await resp.json()

async def main() -> None:
    urls = ["https://..."] * 10
    async with aiohttp.ClientSession() as session:
        # Lance toutes les coroutines "simultanément"
        tasks = [fetch(session, url) for url in urls]
        resultats = await asyncio.gather(*tasks)
    return resultats

# Point d'entrée
asyncio.run(main())

# ✗ Tous les threads reçoivent i=4
threads = []
for i in range(5):
    t = threading.Thread(target=lambda: print(i))
    threads.append(t)

# ✓ Capturer i par valeur
for i in range(5):
    t = threading.Thread(target=lambda n=i: print(n))
    # ou : target=print, args=(i,)

# ✗ Deadlock : ordre différent d'acquisition
def thread_a():
    with lock1:   # acquiert lock1
        with lock2: ...  # attend lock2

def thread_b():
    with lock2:   # acquiert lock2
        with lock1: ...  # attend lock1 → DEADLOCK

# ✓ Même ordre partout
def thread_a():
    with lock1:
        with lock2: ...

def thread_b():
    with lock1:   # même ordre que thread_a
        with lock2: ...