Exemple simple d'utilisation de la file d'attente multitraitement, du pool et du verrouillage

J'ai essayé de lire la documentation sur http://docs.python.org/dev/library/multiprocessing.html mais je suis toujours aux prises avec le multiprocessing Queue, Pool et Locking. Et pour l'instant, j'ai pu construire l'exemple ci-dessous.

En ce qui concerne la file d'attente et le pool, je ne suis pas sûr d'avoir bien compris le concept, donc corrigez-moi si je me trompe. Ce que j'essaie de réaliser, c'est de traiter 2 demandes à la fois (la liste de données en contient 8 dans cet exemple), que dois-je utiliser? Pool pour créer 2 processus capables de gérer deux files d'attente différentes (2 au maximum) ou devrais-je simplement utiliser Queue pour traiter 2 entrées à chaque fois? Le verrou serait d'imprimer correctement les sorties.

import multiprocessing
import time

data = (['a', '2'], ['b', '4'], ['c', '6'], ['d', '8'],
        ['e', '1'], ['f', '3'], ['g', '5'], ['h', '7']
)


def mp_handler(var1):
    for indata in var1:
        p = multiprocessing.Process(target=mp_worker, args=(indata[0], indata[1]))
        p.start()


def mp_worker(inputs, the_time):
    print " Processs %s\tWaiting %s seconds" % (inputs, the_time)
    time.sleep(int(the_time))
    print " Process %s\tDONE" % inputs

if __name__ == '__main__':
    mp_handler(data)

La meilleure solution pour votre problème est d'utiliser un fichier Pool. Utiliser Queues et avoir une fonctionnalité distincte "d'alimentation en file d'attente" est probablement exagéré.

Voici une version légèrement réorganisée de votre programme, cette fois avec seulement 2 processus corallés dans un fichier Pool. Je pense que c'est le moyen le plus simple, avec des modifications minimes du code d'origine:

import multiprocessing
import time

data = (
    ['a', '2'], ['b', '4'], ['c', '6'], ['d', '8'],
    ['e', '1'], ['f', '3'], ['g', '5'], ['h', '7']
)

def mp_worker((inputs, the_time)):
    print " Processs %s\tWaiting %s seconds" % (inputs, the_time)
    time.sleep(int(the_time))
    print " Process %s\tDONE" % inputs

def mp_handler():
    p = multiprocessing.Pool(2)
    p.map(mp_worker, data)

if __name__ == '__main__':
    mp_handler()

Notez que la mp_worker()fonction accepte désormais un seul argument (un tuple des deux arguments précédents) car la map()fonction segmente vos données d'entrée en sous-listes, chaque sous-liste étant donnée comme un argument unique à votre fonction de travail.

Production:

Processs a  Waiting 2 seconds
Processs b  Waiting 4 seconds
Process a   DONE
Processs c  Waiting 6 seconds
Process b   DONE
Processs d  Waiting 8 seconds
Process c   DONE
Processs e  Waiting 1 seconds
Process e   DONE
Processs f  Waiting 3 seconds
Process d   DONE
Processs g  Waiting 5 seconds
Process f   DONE
Processs h  Waiting 7 seconds
Process g   DONE
Process h   DONE

Modifiez selon le commentaire @Thales ci-dessous:

Si vous voulez "un verrou pour chaque limite de pool" afin que vos processus s'exécutent par paires en tandem, ala:

Une attente B en attente | A fait, B fait | C en attente, D en attente | C fait, D fait | ...

puis changez la fonction du gestionnaire pour lancer des pools (de 2 processus) pour chaque paire de données:

def mp_handler():
    subdata = zip(data[0::2], data[1::2])
    for task1, task2 in subdata:
        p = multiprocessing.Pool(2)
        p.map(mp_worker, (task1, task2))

Maintenant, votre sortie est:

 Processs a Waiting 2 seconds
 Processs b Waiting 4 seconds
 Process a  DONE
 Process b  DONE
 Processs c Waiting 6 seconds
 Processs d Waiting 8 seconds
 Process c  DONE
 Process d  DONE
 Processs e Waiting 1 seconds
 Processs f Waiting 3 seconds
 Process e  DONE
 Process f  DONE
 Processs g Waiting 5 seconds
 Processs h Waiting 7 seconds
 Process g  DONE
 Process h  DONE

Cela n'est peut-être pas lié à 100% à la question, mais lors de ma recherche d'un exemple d'utilisation du multitraitement avec une file d'attente, cela apparaît en premier sur Google.

Il s'agit d'un exemple de classe de base que vous pouvez instancier et placer des éléments dans une file d'attente et attendre la fin de la file d'attente. C'est tout ce dont j'avais besoin.

from multiprocessing import JoinableQueue
from multiprocessing.context import Process


class Renderer:
    queue = None

    def __init__(self, nb_workers=2):
        self.queue = JoinableQueue()
        self.processes = [Process(target=self.upload) for i in range(nb_workers)]
        for p in self.processes:
            p.start()

    def render(self, item):
        self.queue.put(item)

    def upload(self):
        while True:
            item = self.queue.get()
            if item is None:
                break

            # process your item here

            self.queue.task_done()

    def terminate(self):
        """ wait until queue is empty and terminate processes """
        self.queue.join()
        for p in self.processes:
            p.terminate()

r = Renderer()
r.render(item1)
r.render(item2)
r.terminate()

Voici mon goto personnel pour ce sujet:

Gist here, (pull requests bienvenus!): Https://gist.github.com/thorsummoner/b5b1dfcff7e7fdd334ec

import multiprocessing
import sys

THREADS = 3

# Used to prevent multiple threads from mixing thier output
GLOBALLOCK = multiprocessing.Lock()


def func_worker(args):
    """This function will be called by each thread.
    This function can not be a class method.
    """
    # Expand list of args into named args.
    str1, str2 = args
    del args

    # Work
    # ...



    # Serial-only Portion
    GLOBALLOCK.acquire()
    print(str1)
    print(str2)
    GLOBALLOCK.release()


def main(argp=None):
    """Multiprocessing Spawn Example
    """
    # Create the number of threads you want
    pool = multiprocessing.Pool(THREADS)

    # Define two jobs, each with two args.
    func_args = [
        ('Hello', 'World',), 
        ('Goodbye', 'World',), 
    ]


    try:
        # Spawn up to 9999999 jobs, I think this is the maximum possible.
        # I do not know what happens if you exceed this.
        pool.map_async(func_worker, func_args).get(9999999)
    except KeyboardInterrupt:
        # Allow ^C to interrupt from any thread.
        sys.stdout.write('\033[0m')
        sys.stdout.write('User Interupt\n')
    pool.close()

if __name__ == '__main__':
    main()

Pour tous ceux qui utilisent des éditeurs comme Komodo Edit (win10), ajoutez sys.stdout.flush()à:

def mp_worker((inputs, the_time)):
    print " Process %s\tWaiting %s seconds" % (inputs, the_time)
    time.sleep(int(the_time))
    print " Process %s\tDONE" % inputs
    sys.stdout.flush()

ou en première ligne pour:

    if __name__ == '__main__':
       sys.stdout.flush()

Cela permet de voir ce qui se passe pendant l'exécution du script; au lieu d'avoir à regarder la boîte de ligne de commande noire.

Voici un exemple de mon code (pour le pool de threads, mais changez simplement le nom de la classe et vous aurez un pool de processus):

def execute_run(rp): 
   ... do something 

pool = ThreadPoolExecutor(6)
for mat in TESTED_MATERIAL:
    for en in TESTED_ENERGIES:
        for ecut in TESTED_E_CUT:
            rp = RunParams(
                simulations, DEST_DIR,
                PARTICLE, mat, 960, 0.125, ecut, en
            )
            pool.submit(execute_run, rp)
pool.join()

Fondamentalement:

• pool = ThreadPoolExecutor(6) crée un pool pour 6 threads
• Ensuite, vous avez un tas de for qui ajoutent des tâches au pool
• pool.submit(execute_run, rp) ajoute une tâche au pool, le premier argument est une fonction appelée dans un thread / processus, le reste des arguments est passé à la fonction appelée.
• pool.join attend que toutes les tâches soient terminées.