Comment sortir proprement d'un script multiprocesseur ?

Question

Je suis en train de construire une application de chat non bloquante pour mon site web, et j'ai décidé d'implémenter du multiprocessing pour gérer l'interrogation de la base de données et la messagerie en temps réel.

Je suppose que lorsqu'un utilisateur clique sur une URL donnée pour voir sa conversation avec l'autre personne, je déclenche le script, le multitraitement commence, les messages sont ajoutés à une file d'attente et affichés sur la page, les nouveaux messages sont envoyés à une file d'attente séparée qui interagit avec la base de données, etc.

Cependant, que se passe-t-il lorsque l'utilisateur quitte cette page ? Je suppose que je dois quitter ces différents processus, mais actuellement, cela ne se prête pas à une sortie "propre". Il faudrait que je mette fin aux processus et, conformément à la documentation sur le multiprocessing :

Warning: If this method (terminate()) is used when the associated process is using a pipe 
or queue then the pipe or queue is liable to become corrupted and may become 
unusable by other process.Similarly, if the process has acquired a lock or 
semaphore etc. then terminating it is liable to cause other processes to 
deadlock.

J'ai également examiné sys.exit() Cependant, il ne quitte pas complètement le script sans l'utilisation de la commande terminate() sur les différents processus.

Voici mon code qui est simplifié pour les besoins de cette question. Si je dois le modifier, c'est tout à fait possible. Je veux simplement m'assurer que je m'y prends correctement.

import multiprocessing
import Queue
import time
import sys

## Get all past messages
def batch_messages():   

    # The messages list here will be attained via a db query
    messages = [">> This is the message.", ">> Hello, how are you doing today?", ">> Really good!"]
    for m in messages:
        print m

## Add messages to the DB   
def add_messages(q2):   

    while True:
        # Retrieve from the queue
        message_to_db = q2.get()

        # For testing purposes only; perfrom another DB query to add the message to the DB
        print message_to_db, "(Add to DB)"

## Recieve new, inputted messages.
def receive_new_message(q1, q2):

    while True:
        # Add the new message to the queue:
        new_message = q1.get()

        # Print the message to the (other user's) screen
        print ">>", new_message

        # Add the q1 message to q2 for databse manipulation
        q2.put(new_message)

def shutdown():
    print "Shutdown initiated"
    p_rec.terminate()
    p_batch.terminate()
    p_add.terminate()
    sys.exit()

if __name__ == "__main__":

    # Set up the queue
    q1 = multiprocessing.Queue()
    q2 = multiprocessing.Queue()

    # Set up the processes
    p_batch = multiprocessing.Process(target=batch_messages)
    p_add = multiprocessing.Process(target=add_messages, args=(q2,))
    p_rec = multiprocessing.Process(target=receive_new_message, args=(q1, q2,))

    # Start the processes   
    p_batch.start() # Perfrom batch get 
    p_rec.start()
    p_add.start()

    time.sleep(0.1) # Test: Sleep to allow proper formatting

    while True:

        # Enter a new message
        input_message = raw_input("Type a message: ")

        # TEST PURPOSES ONLY: shutdown
        if input_message == "shutdown_now":
            shutdown()

        # Add the new message to the queue:
        q1.put(input_message)

        # Let the processes catch up before printing "Type a message: " again. (Shell purposes only)
        time.sleep(0.1)

Comment dois-je faire face à cette situation ? Mon code doit-il être fondamentalement révisé ? et si oui, que dois-je faire pour le corriger ?

Toutes les réflexions, commentaires, révisions ou ressources sont appréciés.

Merci !

Answer 1

Avertissement : je ne connais pas vraiment Python. Mais les concepts de multithreading sont suffisamment similaires dans tous les langages que je connais pour que je me sente suffisamment en confiance pour essayer de répondre quand même.

Lorsque vous utilisez plusieurs threads/processus, chacun d'entre eux doit avoir une étape dans sa boucle pour vérifier une variable (j'appelle souvent la variable "active", ou "keepGoing" ou autre, et c'est généralement un booléen).

Cette variable est généralement soit partagée entre les threads, soit envoyée sous forme de message à chaque thread, en fonction de votre langage de programmation et du moment où vous souhaitez que le processus s'arrête (terminez votre travail d'abord, oui/non ?).

Une fois que la variable est définie, tous les threads quittent leurs boucles de traitement et quittent leurs threads.

Dans votre cas, vous avez une boucle "while true". Celle-ci ne sort jamais. Remplacez-la par "exit" lorsqu'une variable est définie et le thread doit se fermer lorsque la sortie de la fonction est atteinte.