# Fichier: python_cheats/cheatsheets/threading.txt
# Cheatsheet Threading & Concurrence Python - Guide Complet


[OK] CONCEPTS DE BASE

# GIL (Global Interpreter Lock)
# Python a un GIL qui permet qu'un seul thread exécute du bytecode Python à la fois
# [OK] Threading utile pour: I/O-bound (réseau, fichiers, base de données)
# [X] Threading limité pour: CPU-bound (calculs intensifs)
# -> Pour CPU-bound: utiliser multiprocessing

# Quand utiliser Threading:
# - Requêtes HTTP/API
# - Lecture/écriture fichiers
# - Opérations base de données
# - GUI (garder interface responsive)
# - Tâches d'attente (sleep, timers)

# Quand utiliser Multiprocessing:
# - Calculs mathématiques intensifs
# - Traitement d'images/vidéos
# - Machine Learning
# - Compression/décompression


[OK] MODULE THREADING - BASES

import threading
import time

# === Créer et démarrer un thread ===

# Méthode 1: Fonction comme target
def worker():
    print(f"Thread {threading.current_thread().name} démarre")
    time.sleep(2)
    print(f"Thread {threading.current_thread().name} termine")

thread = threading.Thread(target=worker)
thread.start()          # Démarre le thread
thread.join()           # Attend la fin du thread

# Méthode 2: Avec arguments
def worker_with_args(name, duration):
    print(f"Travailleur {name} commence")
    time.sleep(duration)
    print(f"Travailleur {name} termine")

thread = threading.Thread(target=worker_with_args, args=("Alice", 2))
thread.start()
thread.join()

# Méthode 3: Avec arguments nommés
thread = threading.Thread(
    target=worker_with_args,
    kwargs={"name": "Bob", "duration": 3}
)
thread.start()
thread.join()

# Méthode 4: Classe héritant de Thread
class WorkerThread(threading.Thread):
    def __init__(self, name, duration):
        super().__init__()
        self.worker_name = name
        self.duration = duration
    
    def run(self):
        print(f"{self.worker_name} commence")
        time.sleep(self.duration)
        print(f"{self.worker_name} termine")

thread = WorkerThread("Charlie", 2)
thread.start()
thread.join()


[OK] PROPRIÉTÉS ET MÉTHODES DES THREADS

# Nom du thread
thread = threading.Thread(target=worker, name="MonThread")
print(thread.name)
thread.name = "NouveauNom"

# Thread démon
# Démons: threads qui s'arrêtent quand le programme principal termine
thread = threading.Thread(target=worker, daemon=True)
thread.start()
# Ou
thread.daemon = True

# Vérifier si thread est vivant
if thread.is_alive():
    print("Thread en cours d'exécution")

# Identifier le thread
thread_id = thread.ident           # ID du thread
thread_native_id = thread.native_id  # ID natif OS (Python 3.8+)

# Thread courant
current = threading.current_thread()
print(f"Thread actuel: {current.name}")

# Nom du thread principal
main_thread = threading.main_thread()
print(f"Thread principal: {main_thread.name}")

# Compter threads actifs
count = threading.active_count()
print(f"Threads actifs: {count}")

# Lister tous les threads
threads = threading.enumerate()
for t in threads:
    print(f"Thread: {t.name}, Alive: {t.is_alive()}")


[OK] CRÉATION MULTIPLE DE THREADS

import threading

# Créer plusieurs threads
def task(n):
    print(f"Tâche {n} démarre")
    time.sleep(1)
    print(f"Tâche {n} termine")

threads = []
for i in range(5):
    thread = threading.Thread(target=task, args=(i,))
    threads.append(thread)
    thread.start()

# Attendre tous les threads
for thread in threads:
    thread.join()

print("Tous les threads sont terminés")

# Avec compréhension de liste
threads = [threading.Thread(target=task, args=(i,)) for i in range(5)]
for thread in threads:
    thread.start()
for thread in threads:
    thread.join()


[OK] JOIN ET TIMEOUT

import threading
import time

def long_task():
    time.sleep(5)
    print("Tâche longue terminée")

thread = threading.Thread(target=long_task)
thread.start()

# Attendre indéfiniment
thread.join()

# Attendre avec timeout
thread = threading.Thread(target=long_task)
thread.start()
thread.join(timeout=2)  # Attendre max 2 secondes

if thread.is_alive():
    print("Thread encore actif après timeout")
else:
    print("Thread terminé")

# Pattern: Attendre plusieurs threads avec timeout
threads = [threading.Thread(target=long_task) for _ in range(3)]
for t in threads:
    t.start()

timeout = 10
start = time.time()
for t in threads:
    remaining = timeout - (time.time() - start)
    if remaining > 0:
        t.join(timeout=remaining)


[OK] RETOURNER DES VALEURS DEPUIS UN THREAD

# Méthode 1: Variable partagée (nécessite Lock)
import threading

result = []
lock = threading.Lock()

def worker(n):
    res = n * n
    with lock:
        result.append(res)

threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print(result)  # [0, 1, 4, 9, 16]

# Méthode 2: Classe avec attribut
class WorkerThread(threading.Thread):
    def __init__(self, n):
        super().__init__()
        self.n = n
        self.result = None
    
    def run(self):
        self.result = self.n * self.n

threads = [WorkerThread(i) for i in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()

results = [t.result for t in threads]
print(results)  # [0, 1, 4, 9, 16]

# Méthode 3: Queue (recommandé)
from queue import Queue

def worker(n, queue):
    result = n * n
    queue.put(result)

q = Queue()
threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i, q))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

results = []
while not q.empty():
    results.append(q.get())
print(results)


[OK] SYNCHRONISATION - LOCK

import threading

# Problème: race condition sans Lock
counter = 0

def increment():
    global counter
    for _ in range(100000):
        counter += 1  # Pas atomique!

threads = [threading.Thread(target=increment) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(counter)  # Résultat incorrect (< 1000000)

# Solution: Lock
counter = 0
lock = threading.Lock()

def increment_safe():
    global counter
    for _ in range(100000):
        lock.acquire()
        counter += 1
        lock.release()

# Ou avec context manager (recommandé)
def increment_safe():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

threads = [threading.Thread(target=increment_safe) for _ in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(counter)  # 1000000 (correct)

# Méthodes Lock
lock = threading.Lock()

lock.acquire()              # Bloquer jusqu'à obtenir le lock
# ... section critique ...
lock.release()              # Libérer le lock

lock.acquire(blocking=True)    # Attendre le lock
lock.acquire(blocking=False)   # Non-bloquant, retourne False si indisponible
lock.acquire(timeout=5)        # Attendre max 5 secondes

# Vérifier si lock est acquis
is_locked = lock.locked()


[OK] SYNCHRONISATION - RLOCK (Reentrant Lock)

import threading

# Lock normal: deadlock si acquis 2 fois
lock = threading.Lock()

def bad_recursion(n):
    lock.acquire()
    if n > 0:
        bad_recursion(n - 1)  # Deadlock!
    lock.release()

# RLock: peut être acquis plusieurs fois par même thread
rlock = threading.RLock()

def safe_recursion(n):
    rlock.acquire()
    if n > 0:
        safe_recursion(n - 1)  # OK avec RLock
    rlock.release()

# Ou avec context manager
def safe_recursion(n):
    with rlock:
        if n > 0:
            safe_recursion(n - 1)

# Exemple pratique: méthodes s'appelant mutuellement
class BankAccount:
    def __init__(self):
        self._balance = 0
        self._lock = threading.RLock()
    
    def deposit(self, amount):
        with self._lock:
            self._balance += amount
            self._log_transaction("deposit", amount)
    
    def withdraw(self, amount):
        with self._lock:
            self._balance -= amount
            self._log_transaction("withdraw", amount)
    
    def _log_transaction(self, type, amount):
        with self._lock:  # RLock permet ré-acquisition
            print(f"{type}: {amount}, balance: {self._balance}")


[OK] SYNCHRONISATION - SEMAPHORE

import threading
import time

# Limiter nombre de threads concurrents
# Exemple: max 3 connexions simultanées à une ressource

semaphore = threading.Semaphore(3)  # Max 3 threads

def access_resource(n):
    print(f"Thread {n} attend l'accès")
    semaphore.acquire()
    print(f"Thread {n} accède à la ressource")
    time.sleep(2)
    print(f"Thread {n} libère la ressource")
    semaphore.release()

# Avec context manager
def access_resource(n):
    print(f"Thread {n} attend l'accès")
    with semaphore:
        print(f"Thread {n} accède à la ressource")
        time.sleep(2)
        print(f"Thread {n} libère la ressource")

threads = [threading.Thread(target=access_resource, args=(i,)) for i in range(10)]
for t in threads:
    t.start()
for t in threads:
    t.join()

# BoundedSemaphore: empêche release() plus de fois qu'acquire()
bounded_sem = threading.BoundedSemaphore(3)

# Exemple: Pool de connexions base de données
class ConnectionPool:
    def __init__(self, max_connections):
        self.semaphore = threading.Semaphore(max_connections)
        self.connections = []
    
    def acquire_connection(self):
        self.semaphore.acquire()
        # Retourner connection de self.connections
        return connection
    
    def release_connection(self, connection):
        # Remettre connection dans self.connections
        self.semaphore.release()


[OK] SYNCHRONISATION - EVENT

import threading
import time

# Event: signaler entre threads

event = threading.Event()

def waiter():
    print("Waiter: En attente de l'event")
    event.wait()  # Bloquer jusqu'au signal
    print("Waiter: Event reçu!")

def setter():
    print("Setter: Travail en cours...")
    time.sleep(3)
    print("Setter: Envoie l'event")
    event.set()  # Signaler

t1 = threading.Thread(target=waiter)
t2 = threading.Thread(target=setter)

t1.start()
t2.start()
t1.join()
t2.join()

# Méthodes Event
event = threading.Event()

event.set()         # Déclencher l'event
event.clear()       # Réinitialiser l'event
event.is_set()      # Vérifier si déclenché
event.wait()        # Attendre l'event
event.wait(timeout=5)  # Attendre max 5 secondes

# Event avec timeout
event = threading.Event()

def waiter_with_timeout():
    print("En attente...")
    if event.wait(timeout=5):
        print("Event reçu!")
    else:
        print("Timeout!")

# Exemple: Arrêter threads proprement
stop_event = threading.Event()

def worker():
    while not stop_event.is_set():
        print("Travail...")
        time.sleep(1)
    print("Arrêt propre")

thread = threading.Thread(target=worker)
thread.start()

time.sleep(5)
stop_event.set()  # Signaler l'arrêt
thread.join()

# Exemple: Barrière de synchronisation avec Event
ready_event = threading.Event()
start_event = threading.Event()

def racer(n):
    print(f"Coureur {n} prêt")
    ready_event.set()
    start_event.wait()  # Attendre le départ
    print(f"Coureur {n} part!")

threads = [threading.Thread(target=racer, args=(i,)) for i in range(5)]
for t in threads:
    t.start()

ready_event.wait()  # Attendre que tous soient prêts
print("À vos marques... Partez!")
start_event.set()

for t in threads:
    t.join()


[OK] SYNCHRONISATION - CONDITION

import threading
import time

# Condition: Lock + mécanisme de notification

condition = threading.Condition()
items = []

def consumer():
    with condition:
        while not items:
            print("Consumer: en attente d'items")
            condition.wait()  # Libère lock et attend notification
        item = items.pop(0)
        print(f"Consumer: consommé {item}")

def producer():
    time.sleep(2)
    with condition:
        item = "produit"
        items.append(item)
        print(f"Producer: produit {item}")
        condition.notify()  # Notifier un waiter

t1 = threading.Thread(target=consumer)
t2 = threading.Thread(target=producer)
t1.start()
t2.start()
t1.join()
t2.join()

# Méthodes Condition
condition = threading.Condition()

with condition:
    condition.wait()            # Attendre notification
    condition.wait(timeout=5)   # Avec timeout
    condition.notify()          # Notifier 1 thread
    condition.notify(n=3)       # Notifier 3 threads
    condition.notify_all()      # Notifier tous les threads

# Exemple: Producer-Consumer classique
condition = threading.Condition()
buffer = []
MAX_SIZE = 10

def producer(n):
    for i in range(5):
        with condition:
            while len(buffer) >= MAX_SIZE:
                condition.wait()  # Buffer plein
            item = f"item-{n}-{i}"
            buffer.append(item)
            print(f"Producer {n}: produit {item}")
            condition.notify()  # Notifier consumers
        time.sleep(0.5)

def consumer(n):
    for _ in range(5):
        with condition:
            while not buffer:
                condition.wait()  # Buffer vide
            item = buffer.pop(0)
            print(f"Consumer {n}: consommé {item}")
            condition.notify()  # Notifier producers
        time.sleep(1)

producers = [threading.Thread(target=producer, args=(i,)) for i in range(2)]
consumers = [threading.Thread(target=consumer, args=(i,)) for i in range(2)]

for t in producers + consumers:
    t.start()
for t in producers + consumers:
    t.join()


[OK] SYNCHRONISATION - BARRIER

import threading
import time

# Barrier: synchroniser groupe de threads à un point

barrier = threading.Barrier(3)  # Attendre 3 threads

def worker(n):
    print(f"Thread {n} phase 1")
    time.sleep(n)
    
    print(f"Thread {n} attend à la barrière")
    barrier.wait()  # Tous attendent ici
    
    print(f"Thread {n} phase 2 (tous synchronisés)")

threads = [threading.Thread(target=worker, args=(i,)) for i in range(3)]
for t in threads:
    t.start()
for t in threads:
    t.join()

# Barrier avec action
def barrier_action():
    print("=== Tous les threads ont atteint la barrière ===")

barrier = threading.Barrier(3, action=barrier_action)

# Méthodes Barrier
barrier = threading.Barrier(3)

barrier.wait()              # Attendre à la barrière
barrier.wait(timeout=5)     # Avec timeout

barrier.parties             # Nombre de threads requis
barrier.n_waiting           # Nombre de threads en attente

barrier.reset()             # Réinitialiser (lève BrokenBarrierError)
barrier.abort()             # Casser la barrière

# Exemple: Traitement par batch
barrier = threading.Barrier(5)

def process_batch(worker_id, data):
    for batch in data:
        print(f"Worker {worker_id}: traite batch")
        time.sleep(1)
        
        print(f"Worker {worker_id}: attend les autres")
        barrier.wait()  # Synchroniser entre batches
        
        print(f"Worker {worker_id}: batch suivant")


[OK] QUEUE - COMMUNICATION THREAD-SAFE

from queue import Queue, LifoQueue, PriorityQueue
import threading
import time

# Queue FIFO (First In First Out)
q = Queue()

def producer():
    for i in range(5):
        item = f"item-{i}"
        print(f"Producing {item}")
        q.put(item)
        time.sleep(0.5)

def consumer():
    while True:
        item = q.get()
        if item is None:  # Signal d'arrêt
            break
        print(f"Consuming {item}")
        time.sleep(1)
        q.task_done()

t1 = threading.Thread(target=producer)
t2 = threading.Thread(target=consumer)
t1.start()
t2.start()

t1.join()
q.put(None)  # Signal pour arrêter consumer
t2.join()

# Méthodes Queue
q = Queue(maxsize=10)  # Taille max (0 = illimité)

q.put(item)                 # Ajouter item (bloque si pleine)
q.put(item, block=False)    # Non-bloquant (raise Full)
q.put(item, timeout=5)      # Avec timeout

item = q.get()              # Retirer item (bloque si vide)
item = q.get(block=False)   # Non-bloquant (raise Empty)
item = q.get(timeout=5)     # Avec timeout

q.task_done()               # Marquer tâche comme terminée
q.join()                    # Attendre que toutes les tâches soient done

q.qsize()                   # Taille approximative
q.empty()                   # Vérifier si vide
q.full()                    # Vérifier si pleine

# LifoQueue (Last In First Out - Stack)
lifo = LifoQueue()
lifo.put(1)
lifo.put(2)
lifo.put(3)
print(lifo.get())  # 3 (dernier entré)

# PriorityQueue (priorité = plus petit nombre d'abord)
pq = PriorityQueue()
pq.put((3, "tâche basse priorité"))
pq.put((1, "tâche haute priorité"))
pq.put((2, "tâche moyenne priorité"))

print(pq.get())  # (1, "tâche haute priorité")
print(pq.get())  # (2, "tâche moyenne priorité")

# Exemple complet: Producer-Consumer avec Queue
from queue import Queue
import threading
import time
import random

def producer(queue, producer_id):
    for i in range(5):
        item = f"P{producer_id}-Item{i}"
        queue.put(item)
        print(f"Producer {producer_id} produced {item}")
        time.sleep(random.uniform(0.1, 0.5))
    print(f"Producer {producer_id} done")

def consumer(queue, consumer_id):
    while True:
        item = queue.get()
        if item is None:
            queue.task_done()
            break
        print(f"Consumer {consumer_id} consuming {item}")
        time.sleep(random.uniform(0.2, 0.8))
        queue.task_done()
    print(f"Consumer {consumer_id} done")

q = Queue()

# Créer producers et consumers
producers = [threading.Thread(target=producer, args=(q, i)) for i in range(2)]
consumers = [threading.Thread(target=consumer, args=(q, i)) for i in range(3)]

# Démarrer tous
for t in producers + consumers:
    t.start()

# Attendre producers
for t in producers:
    t.join()

# Attendre que la queue soit vide
q.join()

# Arrêter consumers
for _ in consumers:
    q.put(None)
for t in consumers:
    t.join()

print("Tous terminés")


[OK] THREAD POOL - CONCURRENT.FUTURES

from concurrent.futures import ThreadPoolExecutor, as_completed
import time

# ThreadPoolExecutor: pool de threads réutilisables

def task(n):
    print(f"Tâche {n} démarre")
    time.sleep(1)
    return n * n

# Méthode 1: Context manager (recommandé)
with ThreadPoolExecutor(max_workers=5) as executor:
    # submit: soumettre tâche individuelle
    future = executor.submit(task, 10)
    result = future.result()  # Attendre et obtenir résultat
    print(result)  # 100

# Méthode 2: map (comme map() builtin)
with ThreadPoolExecutor(max_workers=5) as executor:
    results = executor.map(task, range(10))
    for result in results:
        print(result)

# submit avec multiple tâches
with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(task, i) for i in range(10)]
    
    # Attendre toutes les futures
    for future in futures:
        result = future.result()
        print(result)

# as_completed: traiter dans l'ordre de complétion
from concurrent.futures import as_completed

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = {executor.submit(task, i): i for i in range(10)}
    
    for future in as_completed(futures):
        original_arg = futures[future]
        result = future.result()
        print(f"Tâche {original_arg} terminée: {result}")

# wait: attendre avec timeout
from concurrent.futures import wait, FIRST_COMPLETED, ALL_COMPLETED

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(task, i) for i in range(10)]
    
    # Attendre toutes
    done, not_done = wait(futures)
    
    # Attendre avec timeout
    done, not_done = wait(futures, timeout=5)
    
    # Attendre première complétée
    done, not_done = wait(futures, return_when=FIRST_COMPLETED)

# Gestion des exceptions
def task_with_error(n):
    if n == 5:
        raise ValueError("Erreur sur 5!")
    return n * n

with ThreadPoolExecutor(max_workers=5) as executor:
    futures = [executor.submit(task_with_error, i) for i in range(10)]
    
    for future in as_completed(futures):
        try:
            result = future.result()
            print(f"Résultat: {result}")
        except Exception as e:
            print(f"Erreur: {e}")

# Callback: fonction appelée quand future complète
def callback(future):
    result = future.result()
    print(f"Callback: résultat = {result}")

with ThreadPoolExecutor(max_workers=5) as executor:
    future = executor.submit(task, 10)
    future.add_done_callback(callback)

# Arrêter pool
executor = ThreadPoolExecutor(max_workers=5)
# ... soumettre tâches ...
executor.shutdown(wait=True)  # Attendre fin des tâches
executor.shutdown(wait=False)  # Ne pas attendre

# Exemple pratique: Téléchargements parallèles
import requests
from concurrent.futures import ThreadPoolExecutor

def download_url(url):
    response = requests.get(url)
    return response.content

urls = [
    "https://example.com/file1",
    "https://example.com/file2",
    "https://example.com/file3",
]

with ThreadPoolExecutor(max_workers=10) as executor:
    results = executor.map(download_url, urls)
    for url, content in zip(urls, results):
        print(f"Downloaded {url}: {len(content)} bytes")


[OK] THREAD LOCAL STORAGE

import threading

# Stockage local à chaque thread
thread_local = threading.local()

def worker(value):
    # Chaque thread a sa propre valeur
    thread_local.data = value
    print(f"Thread {threading.current_thread().name}: {thread_local.data}")

threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

# Exemple: Connexion base de données par thread
import threading

thread_local = threading.local()

def get_connection():
    if not hasattr(thread_local, "connection"):
        thread_local.connection = create_db_connection()
    return thread_local.connection

def worker():
    conn = get_connection()
    # Utiliser connexion
    # Chaque thread a sa propre connexion

# Exemple: Session requests par thread
import threading
import requests

thread_local = threading.local()

def get_session():
    if not hasattr(thread_local, "session"):
        thread_local.session = requests.Session()
    return thread_local.session

def fetch_url(url):
    session = get_session()
    response = session.get(url)
    return response.text


[OK] TIMER

import threading

# Timer: exécuter fonction après délai

def delayed_task():
    print("Tâche exécutée après délai")

timer = threading.Timer(5.0, delayed_task)  # 5 secondes
timer.start()

# Annuler timer avant qu'il s'exécute
timer = threading.Timer(10.0, delayed_task)
timer.start()
time.sleep(2)
timer.cancel()  # Annuler

# Timer avec arguments
def delayed_task_with_args(name, age):
    print(f"Hello {name}, age {age}")

timer = threading.Timer(3.0, delayed_task_with_args, args=("Alice", 30))
timer.start()

# Timer répétitif (manuel)
def repeating_timer(interval, func):
    def wrapper():
        func()
        timer = threading.Timer(interval, wrapper)
        timer.daemon = True
        timer.start()
    wrapper()

def periodic_task():
    print("Tâche périodique")

repeating_timer(5.0, periodic_task)

# Classe pour timer répétitif
class RepeatingTimer:
    def __init__(self, interval, function):
        self.interval = interval
        self.function = function
        self.timer = None
        self.running = False
    
    def _run(self):
        self.running = False
        self.start()
        self.function()
    
    def start(self):
        if not self.running:
            self.timer = threading.Timer(self.interval, self._run)
            self.timer.start()
            self.running = True
    
    def stop(self):
        if self.timer:
            self.timer.cancel()
        self.running = False

timer = RepeatingTimer(2.0, lambda: print("Tick"))
timer.start()
time.sleep(10)
timer.stop()


[OK] PATTERNS COURANTS

# === Pattern 1: Worker Pool ===
from queue import Queue
import threading

def worker(queue):
    while True:
        item = queue.get()
        if item is None:
            break
        process_item(item)
        queue.task_done()

queue = Queue()
threads = []
for i in range(5):
    t = threading.Thread(target=worker, args=(queue,))
    t.start()
    threads.append(t)

# Ajouter tâches
for item in items:
    queue.put(item)

# Attendre complétion
queue.join()

# Arrêter workers
for _ in threads:
    queue.put(None)
for t in threads:
    t.join()

# === Pattern 2: Producer-Consumer ===
# (voir section Queue ci-dessus)

# === Pattern 3: Pipeline ===
from queue import Queue
import threading

def stage1(input_queue, output_queue):
    while True:
        item = input_queue.get()
        if item is None:
            output_queue.put(None)
            break
        result = process_stage1(item)
        output_queue.put(result)
        input_queue.task_done()

def stage2(input_queue, output_queue):
    while True:
        item = input_queue.get()
        if item is None:
            output_queue.put(None)
            break
        result = process_stage2(item)
        output_queue.put(result)
        input_queue.task_done()

queue1 = Queue()
queue2 = Queue()
queue3 = Queue()

t1 = threading.Thread(target=stage1, args=(queue1, queue2))
t2 = threading.Thread(target=stage2, args=(queue2, queue3))
t1.start()
t2.start()

# Ajouter données
for item in data:
    queue1.put(item)

# Signal de fin
queue1.put(None)
queue1.join()
queue2.join()

# Récupérer résultats
results = []
while True:
    item = queue3.get()
    if item is None:
        break
    results.append(item)

t1.join()
t2.join()

# === Pattern 4: Thread avec arrêt propre ===
import threading
import time

class StoppableThread(threading.Thread):
    def __init__(self):
        super().__init__()
        self._stop_event = threading.Event()
    
    def stop(self):
        self._stop_event.set()
    
    def stopped(self):
        return self._stop_event.is_set()
    
    def run(self):
        while not self.stopped():
            # Faire le travail
            print("Travail...")
            time.sleep(1)
        print("Thread arrêté proprement")

thread = StoppableThread()
thread.start()
time.sleep(5)
thread.stop()
thread.join()

# === Pattern 5: Rate Limiting ===
import threading
import time

class RateLimiter:
    def __init__(self, max_calls, time_window):
        self.max_calls = max_calls
        self.time_window = time_window
        self.calls = []
        self.lock = threading.Lock()
    
    def __call__(self, func):
        def wrapper(*args, **kwargs):
            with self.lock:
                now = time.time()
                # Retirer appels anciens
                self.calls = [c for c in self.calls if now - c < self.time_window]
                
                if len(self.calls) >= self.max_calls:
                    sleep_time = self.time_window - (now - self.calls[0])
                    time.sleep(sleep_time)
                    self.calls = self.calls[1:]
                
                self.calls.append(time.time())
            return func(*args, **kwargs)
        return wrapper

@RateLimiter(max_calls=5, time_window=10)
def api_call():
    print(f"API call at {time.time()}")

# === Pattern 6: Singleton Thread-Safe ===
import threading

class Singleton:
    _instance = None
    _lock = threading.Lock()
    
    def __new__(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = super().__new__(cls)
        return cls._instance

# === Pattern 7: Read-Write Lock ===
import threading

class ReadWriteLock:
    def __init__(self):
        self._readers = 0
        self._writers = 0
        self._read_ready = threading.Condition(threading.Lock())
        self._write_ready = threading.Condition(threading.Lock())
    
    def acquire_read(self):
        self._read_ready.acquire()
        while self._writers > 0:
            self._read_ready.wait()
        self._readers += 1
        self._read_ready.release()
    
    def release_read(self):
        self._read_ready.acquire()
        self._readers -= 1
        if self._readers == 0:
            self._read_ready.notify_all()
        self._read_ready.release()
    
    def acquire_write(self):
        self._write_ready.acquire()
        while self._writers > 0 or self._readers > 0:
            self._write_ready.wait()
        self._writers += 1
        self._write_ready.release()
    
    def release_write(self):
        self._write_ready.acquire()
        self._writers -= 1
        self._write_ready.notify_all()
        self._read_ready.acquire()
        self._read_ready.notify_all()
        self._read_ready.release()
        self._write_ready.release()


[OK] DEBUGGING ET MONITORING

import threading
import sys
import traceback

# === Afficher tous les threads ===
def show_threads():
    for thread in threading.enumerate():
        print(f"Thread: {thread.name}")
        print(f"  Daemon: {thread.daemon}")
        print(f"  Alive: {thread.is_alive()}")
        print(f"  Ident: {thread.ident}")

# === Tracer les threads ===
import sys
import threading

def trace_thread():
    current = threading.current_thread()
    frame = sys._current_frames()[current.ident]
    print(f"Thread {current.name}:")
    traceback.print_stack(frame)

# === Détecter deadlocks ===
def detect_deadlock():
    """Afficher les frames de tous les threads"""
    for thread_id, frame in sys._current_frames().items():
        print(f"\nThread {thread_id}:")
        traceback.print_stack(frame)

# === Timeout pour détecter problèmes ===
import signal

def timeout_handler(signum, frame):
    raise TimeoutError("Thread timeout")

def run_with_timeout(func, timeout_duration):
    signal.signal(signal.SIGALRM, timeout_handler)
    signal.alarm(timeout_duration)
    try:
        result = func()
    finally:
        signal.alarm(0)
    return result

# === Logging thread-safe ===
import logging
import threading

# Logger est thread-safe par défaut
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s - %(threadName)s - %(message)s'
)

def worker():
    logging.info("Thread démarre")
    # ...
    logging.info("Thread termine")

# === Profiling threads ===
import cProfile
import pstats

def profile_thread():
    profiler = cProfile.Profile()
    profiler.enable()
    
    # Code à profiler
    worker_function()
    
    profiler.disable()
    stats = pstats.Stats(profiler)
    stats.sort_stats('cumulative')
    stats.print_stats()

# === Exception dans thread ===
def worker():
    try:
        # Code qui peut lever exception
        risky_operation()
    except Exception as e:
        logging.error(f"Erreur dans thread: {e}", exc_info=True)

# Avec ThreadPoolExecutor, exceptions sont capturées
from concurrent.futures import ThreadPoolExecutor

def task_with_exception():
    raise ValueError("Erreur!")

with ThreadPoolExecutor(max_workers=5) as executor:
    future = executor.submit(task_with_exception)
    try:
        result = future.result()
    except ValueError as e:
        print(f"Exception capturée: {e}")


[OK] PERFORMANCE ET OPTIMISATIONS

# === Comparer threading vs séquentiel ===
import time
import threading

def cpu_bound_task(n):
    """Calcul intensif - threading PAS optimal"""
    result = 0
    for i in range(n):
        result += i ** 2
    return result

def io_bound_task(duration):
    """I/O - threading OPTIMAL"""
    time.sleep(duration)
    return "Done"

# Test CPU-bound
start = time.time()
results = [cpu_bound_task(1000000) for _ in range(4)]
print(f"Séquentiel CPU: {time.time() - start:.2f}s")

start = time.time()
threads = [threading.Thread(target=cpu_bound_task, args=(1000000,)) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"Threading CPU: {time.time() - start:.2f}s")  # Plus lent!

# Test I/O-bound
start = time.time()
results = [io_bound_task(1) for _ in range(4)]
print(f"Séquentiel I/O: {time.time() - start:.2f}s")  # ~4s

start = time.time()
threads = [threading.Thread(target=io_bound_task, args=(1,)) for _ in range(4)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(f"Threading I/O: {time.time() - start:.2f}s")  # ~1s, beaucoup mieux!

# === Choisir nombre de threads ===
import os

# CPU-bound: nombre de cores
num_cores = os.cpu_count()
print(f"Cores disponibles: {num_cores}")

# I/O-bound: dépend du cas
# Réseau/API: 10-100+ threads
# Fichiers: 2-10 threads
# Base de données: pool de connexions (5-20)

# === Réduire overhead ===
# Réutiliser threads avec ThreadPoolExecutor
from concurrent.futures import ThreadPoolExecutor

with ThreadPoolExecutor(max_workers=10) as executor:
    # Pool réutilise threads
    for i in range(1000):
        executor.submit(task, i)

# Éviter trop de locks
# Mauvais: lock pour chaque opération
lock = threading.Lock()
for item in items:
    with lock:
        process(item)

# Mieux: batch les opérations
with lock:
    for item in items:
        process(item)

# === Thread pool sizing ===
# Formule de Little: optimal = latency × throughput
# Exemple: 100ms latency, 50 req/s -> 5 threads

def calculate_thread_pool_size(avg_latency_sec, target_throughput):
    return int(avg_latency_sec * target_throughput) + 1


[OK] TESTING

import unittest
import threading
import time

class TestThreading(unittest.TestCase):
    
    def test_basic_thread(self):
        """Test création et exécution thread"""
        result = []
        
        def worker():
            result.append(42)
        
        thread = threading.Thread(target=worker)
        thread.start()
        thread.join()
        
        self.assertEqual(result[0], 42)
    
    def test_thread_synchronization(self):
        """Test synchronisation avec Lock"""
        counter = 0
        lock = threading.Lock()
        
        def increment():
            nonlocal counter
            for _ in range(1000):
                with lock:
                    counter += 1
        
        threads = [threading.Thread(target=increment) for _ in range(10)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        
        self.assertEqual(counter, 10000)
    
    def test_queue_communication(self):
        """Test communication via Queue"""
        from queue import Queue
        
        q = Queue()
        
        def producer():
            for i in range(5):
                q.put(i)
        
        def consumer():
            results = []
            for _ in range(5):
                results.append(q.get())
            return results
        
        t1 = threading.Thread(target=producer)
        t1.start()
        t1.join()
        
        results = consumer()
        self.assertEqual(results, [0, 1, 2, 3, 4])
    
    def test_timeout(self):
        """Test avec timeout"""
        def long_task():
            time.sleep(5)
        
        thread = threading.Thread(target=long_task)
        thread.start()
        thread.join(timeout=1)
        
        self.assertTrue(thread.is_alive())
    
    def test_thread_local(self):
        """Test thread local storage"""
        thread_local = threading.local()
        results = {}
        
        def worker(value):
            thread_local.data = value
            time.sleep(0.1)
            results[threading.current_thread().name] = thread_local.data
        
        threads = [threading.Thread(target=worker, args=(i,), name=f"T{i}") 
                   for i in range(5)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        
        for i in range(5):
            self.assertEqual(results[f"T{i}"], i)

# Test race condition
class TestRaceCondition(unittest.TestCase):
    
    def test_race_condition_without_lock(self):
        """Démontre race condition"""
        counter = 0
        
        def increment():
            nonlocal counter
            for _ in range(10000):
                counter += 1
        
        threads = [threading.Thread(target=increment) for _ in range(10)]
        for t in threads:
            t.start()
        for t in threads:
            t.join()
        
        # Sans lock, résultat incorrect
        self.assertLess(counter, 100000)


[OK] ALTERNATIVES AU THREADING

# === MULTIPROCESSING (CPU-bound) ===
from multiprocessing import Process, Pool, Queue, Lock

def cpu_intensive_task(n):
    result = sum(i*i for i in range(n))
    return result

# Avec Process
if __name__ == '__main__':
    processes = []
    for i in range(4):
        p = Process(target=cpu_intensive_task, args=(1000000,))
        processes.append(p)
        p.start()
    
    for p in processes:
        p.join()

# Avec Pool
if __name__ == '__main__':
    with Pool(processes=4) as pool:
        results = pool.map(cpu_intensive_task, [1000000] * 4)

# === ASYNCIO (I/O-bound) ===
import asyncio

async def async_task(n):
    print(f"Task {n} start")
    await asyncio.sleep(1)
    print(f"Task {n} end")
    return n * 2

async def main():
    # Exécuter plusieurs tâches concurrentes
    tasks = [async_task(i) for i in range(5)]
    results = await asyncio.gather(*tasks)
    print(results)

asyncio.run(main())

# === COMPARAISON ===
# Threading:
# [OK] I/O-bound (réseau, fichiers, DB)
# [OK] Partage mémoire facile
# [X] GIL limite CPU-bound
# [X] Race conditions possibles

# Multiprocessing:
# [OK] CPU-bound (calculs intensifs)
# [OK] Vraie parallélisation
# [X] Overhead important (création process)
# [X] Partage mémoire complexe

# Asyncio:
# [OK] I/O-bound (très efficace)
# [OK] Pas de race conditions
# [OK] Léger (pas de threads OS)
# [X] Nécessite code async/await
# [X] Bloque si opération CPU intensive


[OK] EXEMPLES PRATIQUES

# === Exemple 1: Web Scraper ===
import threading
import requests
from queue import Queue

def scrape_url(url, results):
    try:
        response = requests.get(url, timeout=10)
        results.append({
            'url': url,
            'status': response.status_code,
            'length': len(response.content)
        })
    except Exception as e:
        results.append({'url': url, 'error': str(e)})

urls = [
    "https://example.com",
    "https://google.com",
    "https://github.com",
]

results = []
lock = threading.Lock()

def worker(url):
    response = scrape_url(url, results)
    with lock:
        print(f"Scraped {url}")

threads = [threading.Thread(target=worker, args=(url,)) for url in urls]
for t in threads:
    t.start()
for t in threads:
    t.join()

print(f"Total scraped: {len(results)}")

# === Exemple 2: Batch Processing ===
from concurrent.futures import ThreadPoolExecutor

def process_file(filename):
    with open(filename, 'r') as f:
        data = f.read()
    # Traiter data
    processed = data.upper()
    output_filename = f"processed_{filename}"
    with open(output_filename, 'w') as f:
        f.write(processed)
    return output_filename

files = ['file1.txt', 'file2.txt', 'file3.txt']

with ThreadPoolExecutor(max_workers=5) as executor:
    results = executor.map(process_file, files)
    for result in results:
        print(f"Processed: {result}")

# === Exemple 3: Progress Bar avec Threads ===
import threading
import time

def long_task(task_id, progress):
    for i in range(100):
        time.sleep(0.05)
        progress[task_id] = i + 1

progress = {i: 0 for i in range(5)}
threads = [threading.Thread(target=long_task, args=(i, progress)) 
           for i in range(5)]

for t in threads:
    t.start()

# Afficher progression
while any(t.is_alive() for t in threads):
    for task_id, pct in progress.items():
        print(f"Task {task_id}: {pct}%", end=" | ")
    print("\r", end="")
    time.sleep(0.1)

for t in threads:
    t.join()

print("\nToutes les tâches terminées!")

# === Exemple 4: Server avec Thread par Client ===
import socket
import threading

def handle_client(client_socket, address):
    print(f"Connexion de {address}")
    try:
        while True:
            data = client_socket.recv(1024)
            if not data:
                break
            client_socket.send(data.upper())
    finally:
        client_socket.close()
        print(f"Déconnexion de {address}")

def run_server(host='localhost', port=9999):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.bind((host, port))
    server.listen(5)
    print(f"Serveur écoute sur {host}:{port}")
    
    try:
        while True:
            client_socket, address = server.accept()
            client_thread = threading.Thread(
                target=handle_client,
                args=(client_socket, address)
            )
            client_thread.daemon = True
            client_thread.start()
    except KeyboardInterrupt:
        print("\nServeur arrêté")
    finally:
        server.close()

# === Exemple 5: Cache Thread-Safe ===
import threading
import time

class ThreadSafeCache:
    def __init__(self, ttl=60):
        self._cache = {}
        self._lock = threading.RLock()
        self._ttl = ttl
    
    def get(self, key):
        with self._lock:
            if key in self._cache:
                value, timestamp = self._cache[key]
                if time.time() - timestamp < self._ttl:
                    return value
                else:
                    del self._cache[key]
            return None
    
    def set(self, key, value):
        with self._lock:
            self._cache[key] = (value, time.time())
    
    def clear(self):
        with self._lock:
            self._cache.clear()

cache = ThreadSafeCache(ttl=10)

def worker(worker_id):
    # Lire du cache
    value = cache.get('shared_data')
    if value is None:
        # Calculer si pas en cache
        value = expensive_computation()
        cache.set('shared_data', value)
    print(f"Worker {worker_id}: {value}")

# === Exemple 6: Background Task ===
import threading
import time

class BackgroundTask:
    def __init__(self, interval, function):
        self.interval = interval
        self.function = function
        self.thread = None
        self.stop_event = threading.Event()
    
    def _run(self):
        while not self.stop_event.wait(self.interval):
            self.function()
    
    def start(self):
        if self.thread is None or not self.thread.is_alive():
            self.stop_event.clear()
            self.thread = threading.Thread(target=self._run)
            self.thread.daemon = True
            self.thread.start()
    
    def stop(self):
        self.stop_event.set()
        if self.thread:
            self.thread.join()

def cleanup_task():
    print("Nettoyage périodique...")
    # Nettoyer fichiers temporaires, logs, etc.

task = BackgroundTask(interval=60, function=cleanup_task)
task.start()

# Plus tard...
task.stop()


[OK] BONNES PRATIQUES

# 1. Toujours utiliser context managers
# [OK] Bon
with lock:
    # Code protégé
    pass

# [X] Mauvais
lock.acquire()
# Code protégé
lock.release()  # Peut être oublié si exception!

# 2. Éviter variables globales
# [X] Mauvais
counter = 0

def worker():
    global counter
    counter += 1

# [OK] Bon
def worker(counter_obj):
    counter_obj.increment()

# 3. Utiliser Queue pour communication
# [OK] Queue est thread-safe
# [X] Listes/dicts nécessitent Lock

# 4. Nommer threads
thread = threading.Thread(target=worker, name="WorkerThread1")

# 5. Utiliser daemon threads pour tâches background
thread = threading.Thread(target=background_task, daemon=True)

# 6. Toujours join() les threads non-daemon
for thread in threads:
    thread.join()

# 7. Gérer exceptions dans threads
def safe_worker():
    try:
        risky_operation()
    except Exception as e:
        logging.error(f"Erreur: {e}")

# 8. Préférer ThreadPoolExecutor à Thread manuel
# [OK] Gestion automatique du pool
# [OK] Gestion exceptions
# [OK] Plus propre

# 9. Éviter trop de locks (deadlocks)
# Toujours acquérir locks dans le même ordre

# 10. Documenter comportement thread
def worker():
    """
    Worker thread qui traite les tâches.
    Thread-safe, utilise queue pour communication.
    """
    pass


[OK] PIÈGES COURANTS

# === Piège 1: Variables mutables partagées ===
# Problème
shared_list = []

def worker():
    shared_list.append(1)  # Race condition!

# Solution
lock = threading.Lock()

def worker():
    with lock:
        shared_list.append(1)

# === Piège 2: Oublier join() ===
# Problème
thread = threading.Thread(target=worker)
thread.start()
# Programme termine avant thread!

# Solution
thread.start()
thread.join()  # Attendre

# === Piège 3: Deadlock ===
# Problème
lock1 = threading.Lock()
lock2 = threading.Lock()

def worker1():
    with lock1:
        time.sleep(0.1)
        with lock2:  # Deadlock si worker2 a lock2!
            pass

def worker2():
    with lock2:
        time.sleep(0.1)
        with lock1:  # Deadlock si worker1 a lock1!
            pass

# Solution: Toujours même ordre
def worker1():
    with lock1:
        with lock2:
            pass

def worker2():
    with lock1:  # Même ordre!
        with lock2:
            pass

# === Piège 4: GIL pour CPU-bound ===
# Threading PAS efficace pour calculs
# Utiliser multiprocessing à la place

# === Piège 5: Exceptions silencieuses ===
# Exceptions dans threads n'arrêtent pas programme
def worker():
    raise ValueError("Erreur!")  # Silencieuse!

# Toujours gérer exceptions

# === Piège 6: Utiliser Lock au lieu de RLock ===
# Lock ne peut pas être ré-acquis par même thread
lock = threading.Lock()

def recursive(n):
    with lock:
        if n > 0:
            recursive(n - 1)  # Deadlock!

# Utiliser RLock pour récursion

# === Piège 7: Race condition sur is_alive() ===
if not thread.is_alive():
    thread.start()  # Race: peut être démarré entre les 2!

# Mieux: try/except
try:
    thread.start()
except RuntimeError:
    pass  # Déjà démarré


[OK] RESSOURCES

# Documentation officielle:
# threading: https://docs.python.org/3/library/threading.html
# concurrent.futures: https://docs.python.org/3/library/concurrent.futures.html
# queue: https://docs.python.org/3/library/queue.html
# multiprocessing: https://docs.python.org/3/library/multiprocessing.html

# Guides:
# Real Python - Threading
# https://realpython.com/intro-to-python-threading/

# Real Python - Speed Up Python with Concurrency
# https://realpython.com/python-concurrency/

# Livres:
# - "Python Concurrency with asyncio" par Matthew Fowler
# - "High Performance Python" par Micha Gorelick

# Outils utiles:
# - py-spy: profiler pour Python
# - threading-debug: aide au debugging threads