Ejemplo muy simple de usar la cola de multiprocesamiento, el grupo y el bloqueo

Intenté leer la documentación en http://docs.python.org/dev/library/multiprocessing.html pero todavía estoy luchando con el multiprocesamiento de cola, agrupación y bloqueo. Y por ahora pude construir el siguiente ejemplo.

Con respecto a la cola y la agrupación, no estoy seguro de haber entendido el concepto correctamente, así que corrígeme si me equivoco. Lo que estoy tratando de lograr es procesar 2 solicitudes a la vez (la lista de datos tiene 8 en este ejemplo), entonces, ¿qué debo usar? Pool para crear 2 procesos que pueden manejar dos colas diferentes (2 como máximo) o debería usar Queue para procesar 2 entradas cada vez? El bloqueo sería imprimir las salidas correctamente.

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 mejor solución para su problema es utilizar un Pool. Usar Queuesy tener una funcionalidad separada de "alimentación en cola" probablemente sea excesivo.

Aquí hay una versión ligeramente reorganizada de su programa, esta vez con solo 2 procesos agrupados en un Pool. Creo que es la forma más fácil de hacerlo, con cambios mínimos en el código original:

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()

Tenga en cuenta que la mp_worker()función ahora acepta un solo argumento (una tupla de los dos argumentos anteriores) porque la map()función fragmenta sus datos de entrada en sublistas, cada sublista se proporciona como un solo argumento para su función de trabajo.

Salida:

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

Edite según el comentario de @Thales a continuación:

Si desea "un bloqueo para cada límite de grupo" para que sus procesos se ejecuten en pares en tándem, ala:

A esperando B esperando | A hecho, B hecho | C esperando, D esperando | C hecho, D hecho | ...

luego cambie la función del controlador para lanzar grupos (de 2 procesos) para cada par de datos:

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))

Ahora tu salida es:

 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

Esto puede no estar 100% relacionado con la pregunta, pero en mi búsqueda de un ejemplo de uso de multiprocesamiento con una cola, esto aparece primero en Google.

Esta es una clase de ejemplo básica que puede crear instancias y poner elementos en una cola y puede esperar hasta que la cola finalice. Eso es todo lo que necesitaba.

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()

Aquí está mi goto personal para este tema:

Gist aquí, (¡solicitudes de extracción son bienvenidas!): 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()

Para todos los que usan editores como Komodo Edit (win10), agregue sys.stdout.flush()a:

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()

o como primera línea para:

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

Esto ayuda a ver lo que sucede durante la ejecución del script; en lugar de tener que mirar el cuadro de línea de comando negro.

Aquí hay un ejemplo de mi código (para el grupo de subprocesos, pero simplemente cambie el nombre de la clase y tendrá el grupo de procesos):

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()

Básicamente:

• pool = ThreadPoolExecutor(6) crea un grupo de 6 hilos
• Entonces tienes un montón de for's que agregan tareas al grupo
• pool.submit(execute_run, rp) agrega una tarea al grupo, el primer argumento es una función llamada en un hilo / proceso, el resto de los argumentos se pasan a la función llamada.
• pool.join espera hasta que se completen todas las tareas.