Introduzione
Quando un programma usa il multithreading o il multiprocessing, interrompere un’attività non significa semplicemente terminare il thread o il processo. Un arresto forzato può lasciare file aperti, connessioni di rete attive, dati parzialmente elaborati o risorse non rilasciate.
Una soluzione più affidabile consiste nell’implementare un arresto cooperativo: il programma segnala ai lavoratori che devono fermarsi e ogni lavoratore decide autonomamente quando interrompere il proprio ciclo, completando prima le operazioni indispensabili.
In questo tutorial useremo:
- threading.Event per segnalare l’arresto a un thread;
- multiprocessing.Event per condividere la segnalazione tra processi;
- un thread supervisore che decide quando interrompere l’elaborazione;
- la gestione ordinata della chiusura tramite
join(); - il modulo
loggingper osservare il comportamento dei lavoratori.
Questo approccio è utile nei crawler, nei servizi che elaborano file, nei programmi batch e nelle applicazioni che devono reagire a un comando di arresto senza perdere il lavoro già iniziato.
Codice completo
import logging
import multiprocessing as mp
import threading
import time
def configura_logging():
"""Configura un formato comune per tutti i messaggi."""
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(processName)s | "
"%(threadName)s | %(message)s",
datefmt="%H:%M:%S",
)
def lavoratore_thread(stop_event):
"""Esegue un lavoro nel thread principale del processo."""
logger = logging.getLogger("thread-worker")
for numero in range(1, 21):
# wait() restituisce True se l´evento è stato attivato.
# In questo modo il thread non resta bloccato troppo a lungo.
if stop_event.wait(timeout=0.3):
logger.info("Arresto richiesto: il thread termina.")
return
logger.info("Thread: elaborazione elemento %d", numero)
logger.info("Thread: lavoro completato.")
def lavoratore_processo(numero_processo, stop_event):
"""Esegue un lavoro indipendente in un processo separato."""
configura_logging()
logger = logging.getLogger(f"process-{numero_processo}")
for numero in range(1, 21):
if stop_event.wait(timeout=0.4):
logger.info(
"Arresto richiesto: il processo termina dopo "
"l´elemento %d.",
numero - 1,
)
return
logger.info(
"Processo %d: elaborazione elemento %d",
numero_processo,
numero,
)
logger.info("Processo %d: lavoro completato.", numero_processo)
def supervisore(stop_event, secondi):
"""Attende un intervallo e richiede l´arresto."""
logger = logging.getLogger("supervisore")
time.sleep(secondi)
logger.info("Tempo massimo raggiunto: richiesta di arresto.")
stop_event.set()
def main():
configura_logging()
# Event per il thread locale.
stop_thread = threading.Event()
# Event condivisibile tra processi.
stop_processi = mp.Event()
thread = threading.Thread(
target=lavoratore_thread,
args=(stop_thread,),
name="ThreadLavoratore",
)
processi = [
mp.Process(
target=lavoratore_processo,
args=(indice, stop_processi),
name=f"ProcessoLavoratore-{indice}",
)
for indice in range(1, 3)
]
# Il supervisore è un thread perché deve coordinare il programma
# principale senza creare un ulteriore processo.
supervisore_thread = threading.Thread(
target=supervisore,
args=(stop_thread, 2.0),
name="SupervisoreThread",
)
thread.start()
supervisore_thread.start()
for processo in processi:
processo.start()
# Quando il thread locale termina, notifichiamo l´arresto anche
# ai processi. In un´applicazione reale questa decisione potrebbe
# dipendere da un segnale, da un errore o da un comando dell´utente.
thread.join()
stop_processi.set()
for processo in processi:
processo.join(timeout=3)
if processo.is_alive():
logging.warning(
"%s non ha terminato in tempo: terminazione forzata.",
processo.name,
)
processo.terminate()
processo.join()
supervisore_thread.join()
logging.info("Tutte le attività sono state chiuse.")
if __name__ == "__main__":
# Necessario soprattutto su Windows e macOS con il metodo spawn.
mp.freeze_support()
main()
Spiegazione
Perché usare un evento?
Un oggetto Event rappresenta uno stato booleano condiviso. Inizialmente l’evento non è attivo. Quando viene chiamato set(), l’evento passa allo stato attivo e tutti i lavoratori possono rilevarlo con is_set() oppure con wait().
Nel codice il thread controlla l’evento tramite:
if stop_event.wait(timeout=0.3):
return
Questa forma è spesso migliore di una semplice chiamata a time.sleep(). Se l’evento viene attivato, wait() ritorna subito, evitando di attendere inutilmente la fine del timeout.
Thread e processi richiedono eventi diversi
threading.Event è progettato per la condivisione dello stato tra thread dello stesso processo. Non deve essere usato per coordinare processi separati.
Per i processi si usa invece multiprocessing.Event. Questo oggetto utilizza primitive compatibili con la memoria condivisa tra processi e può essere passato alla funzione eseguita da ciascun processo.
Nel programma sono presenti quindi due eventi:
stop_thread, usato dal thread lavoratore e dal supervisore;stop_processi, usato dal processo principale e dai processi lavoratori.
Chiusura ordinata con join
La chiamata join() blocca il programma chiamante fino alla terminazione del thread o del processo. È importante usarla per evitare lavoratori lasciati in esecuzione dopo la fine della funzione main().
Per i processi è stato aggiunto un timeout:
processo.join(timeout=3)
if processo.is_alive():
processo.terminate()
La terminazione forzata è una misura di emergenza, non il meccanismo principale. Deve essere usata soltanto quando un processo non risponde all’arresto cooperativo, perché può interrompere un’operazione a metà.
Il ruolo del supervisore
Il thread supervisore simula un componente esterno che richiede la chiusura del programma dopo un certo tempo. In un’applicazione reale, lo stesso comportamento potrebbe essere attivato da:
- un segnale del sistema operativo;
- un pulsante “Stop” in un’interfaccia grafica;
- un comando ricevuto tramite rete;
- un errore non recuperabile;
- il superamento di un limite di tempo.
Best practice
- Controllare spesso l’evento: un lavoratore che esegue una singola operazione di dieci minuti non può reagire rapidamente all’arresto. Suddividere il lavoro in unità più piccole rende il programma più responsivo.
- Preferire
wait(timeout)a cicli con attese lunghe: permette di interrompere rapidamente l’attesa quando arriva la richiesta di arresto. - Chiudere sempre le risorse: file, socket e connessioni al database devono essere gestiti con
witho con blocchitry/finally. - Usare
logginginvece diprint(): i log includono processo e thread, rendendo più semplice analizzare problemi di concorrenza. - Proteggere il punto di ingresso: il blocco
if __name__ == "__main__"è indispensabile quando si usano processi, soprattutto con il metodospawn. - Non abusare di
terminate(): la chiusura forzata può lasciare dati inconsistenti e risorse in uno stato imprevedibile. - Definire timeout realistici: un timeout troppo breve può causare terminazioni premature; uno troppo lungo ritarda la chiusura dell’applicazione.
Riepilogo
Un sistema multithread o multiprocesso robusto deve prevedere anche come terminare le attività. L’arresto cooperativo con Event permette ai lavoratori di interrompersi in modo controllato, riducendo il rischio di perdita dati e risorse non rilasciate.
Per i thread si usa threading.Event, mentre per i processi si usa multiprocessing.Event. La sequenza consigliata è: segnalare l’arresto, attendere con join(), verificare eventuali blocchi e ricorrere alla terminazione forzata solo come ultima risorsa.