Ejemplo de Multiprocesamiento en Python

En nuestro tutorial anterior, aprendimos sobre Ejemplo de CSV en Python. En este tutorial vamos a aprender sobre Multiprocesamiento en Python con ejemplos.

Multiprocesamiento en Python

El procesamiento paralelo está recibiendo más atención en la actualidad. Si aún no sabe sobre el procesamiento paralelo, aprenda de Wikipedia. A medida que los fabricantes de CPU continúan agregando más y más núcleos a sus procesadores, crear código paralelo es una excelente manera de mejorar el rendimiento. Python introdujo el módulo multiprocessing para permitirnos escribir código paralelo. Para comprender la principal motivación de este módulo, debemos conocer algunos conceptos básicos sobre programación paralela. Después de leer este artículo, esperamos que puedas reunir algunos conocimientos sobre este tema.

Multiprocesamiento en Python: Proceso, Cola y Bloqueos

Hay muchas clases en el módulo de multiprocesamiento de Python para construir un programa paralelo. Entre ellas, tres clases básicas son Process, Queue y Lock. Estas clases te ayudarán a construir un programa paralelo. Pero antes de describir sobre ellas, iniciemos este tema con un código simple. Para que un programa paralelo sea útil, debes saber cuántos núcleos hay en tu PC. El módulo de Multiprocesamiento de Python te permite saber eso. El siguiente código simple imprimirá el número de núcleos en tu PC.

import multiprocessing

print("Number of cpu : ", multiprocessing.cpu_count())

La siguiente salida puede variar para tu PC. Para mí, el número de núcleos es 8.

Clase Process de multiprocessing en Python

La clase Process de multiprocesamiento de Python es una abstracción que configura otro proceso de Python, proporciona código para ejecutar y una forma para que la aplicación principal controle la ejecución. Hay dos funciones importantes que pertenecen a la clase Process: la función start() y la función join(). Primero, necesitamos escribir una función que será ejecutada por el proceso. Luego, necesitamos instanciar un objeto de proceso. Si creamos un objeto de proceso, no sucederá nada hasta que le indiquemos que comience a procesar mediante la función start(). Entonces, el proceso se ejecutará y devolverá su resultado. Después de eso, indicamos al proceso que complete su ejecución mediante la función join(). Sin la llamada a la función join(), el proceso permanecerá inactivo y no terminará. Por lo tanto, si creamos muchos procesos y no los terminamos, podemos enfrentarnos a la escasez de recursos. Entonces, es posible que necesitemos terminarlos manualmente. Una cosa importante es que, si queremos pasar algún argumento a través del proceso, necesitamos usar el argumento de palabra clave args. El siguiente código será útil para entender el uso de la clase Process.

from multiprocessing import Process


def print_func(continent='Asia'):
    print('The name of continent is : ', continent)

if __name__ == "__main__":  # confirms that the code is under main function
    names = ['America', 'Europe', 'Africa']
    procs = []
    proc = Process(target=print_func)  # instantiating without any argument
    procs.append(proc)
    proc.start()

    # instanciando proceso con argumentos
    for name in names:
        # print(name)
        proc = Process(target=print_func, args=(name,))
        procs.append(proc)
        proc.start()

    # completar los procesos
    for proc in procs:
        proc.join()

La salida del siguiente código será:

Clase Queue de multiprocesamiento en Python

Tienes conocimientos básicos sobre estructuras de datos informáticas, probablemente sabes sobre Cola. El módulo de multiprocesamiento de Python proporciona la clase Queue, que es exactamente una estructura de datos Primero en entrar, primero en salir (FIFO). Pueden almacenar cualquier objeto de Python pickle (aunque los más simples son los mejores) y son extremadamente útiles para compartir datos entre procesos. Las colas son especialmente útiles cuando se pasan como parámetro a la función objetivo de un proceso para permitir que el proceso consuma datos. Mediante el uso de la función put() podemos insertar datos en la cola y usando get() podemos obtener elementos de las colas. Mira el siguiente código para un ejemplo rápido.

from multiprocessing import Queue

colors = ['red', 'green', 'blue', 'black']
cnt = 1
# instanciando un objeto de cola
queue = Queue()
print('pushing items to queue:')
for color in colors:
    print('item no: ', cnt, ' ', color)
    queue.put(color)
    cnt += 1

print('\npopping items from queue:')
cnt = 0
while not queue.empty():
    print('item no: ', cnt, ' ', queue.get())
    cnt += 1

Clase de Bloqueo de Multiprocesamiento de Python

La tarea de la clase Lock es bastante simple. Permite que el código reclame un bloqueo para que ningún otro proceso pueda ejecutar el código similar hasta que se haya liberado el bloqueo. Entonces, la tarea de la clase Lock es principalmente dos. Uno es reclamar el bloqueo y el otro es liberar el bloqueo. Para reclamar el bloqueo, se utiliza la función acquire() y para liberar el bloqueo se utiliza la función release().

Ejemplo de multiprocesamiento de Python

En este ejemplo de multiprocesamiento en Python, vamos a fusionar todo nuestro conocimiento juntos. Supongamos que tenemos algunas tareas que cumplir. Para completar esa tarea, utilizaremos varios procesos. Por lo tanto, mantendremos dos colas. Una contendrá las tareas y la otra contendrá el registro de las tareas completadas. Luego instanciamos los procesos para completar la tarea. Tenga en cuenta que la clase Queue de Python ya está sincronizada. Eso significa que no necesitamos usar la clase Lock para bloquear múltiples procesos para acceder al mismo objeto de cola. Por eso, no necesitamos usar la clase Lock en este caso. A continuación se muestra la implementación donde estamos agregando tareas a la cola, luego creando procesos y comenzándolos, luego usando join() para completar los procesos. Finalmente, estamos imprimiendo el registro desde la segunda cola.

from multiprocessing import Lock, Process, Queue, current_process
import time
import queue # imported for using queue.Empty exception


def do_job(tasks_to_accomplish, tasks_that_are_done):
    while True:
        try:
            '''
                try to get task from the queue. get_nowait() function will 
                raise queue.Empty exception if the queue is empty. 
                queue(False) function would do the same task also.
            '''
            task = tasks_to_accomplish.get_nowait()
        except queue.Empty:

            break
        else:
            '''
                if no exception has been raised, add the task completion 
                message to task_that_are_done queue
            '''
            print(task)
            tasks_that_are_done.put(task + ' is done by ' + current_process().name)
            time.sleep(.5)
    return True


def main():
    number_of_task = 10
    number_of_processes = 4
    tasks_to_accomplish = Queue()
    tasks_that_are_done = Queue()
    processes = []

    for i in range(number_of_task):
        tasks_to_accomplish.put("Task no " + str(i))

    # creando procesos
    for w in range(number_of_processes):
        p = Process(target=do_job, args=(tasks_to_accomplish, tasks_that_are_done))
        processes.append(p)
        p.start()

    # completando proceso
    for p in processes:
        p.join()

    # imprimir la salida
    while not tasks_that_are_done.empty():
        print(tasks_that_are_done.get())

    return True


if __name__ == '__main__':
    main()

Dependiendo del número de tareas, el código tomará algún tiempo para mostrarle la salida. La salida del siguiente código variará de vez en cuando.

Python multiprocessing Pool

Python multiprocessing Pool se puede utilizar para la ejecución paralela de una función en múltiples valores de entrada, distribuyendo los datos de entrada entre los procesos (paralelismo de datos). A continuación se muestra un ejemplo simple de Python multiprocessing Pool.

from multiprocessing import Pool

import time

work = (["A", 5], ["B", 2], ["C", 1], ["D", 3])


def work_log(work_data):
    print(" Process %s waiting %s seconds" % (work_data[0], work_data[1]))
    time.sleep(int(work_data[1]))
    print(" Process %s Finished." % work_data[0])


def pool_handler():
    p = Pool(2)
    p.map(work_log, work)


if __name__ == '__main__':
    pool_handler()

La siguiente imagen muestra la salida del programa anterior. Observa que el tamaño del grupo es 2, por lo que dos ejecuciones de la función work_log están ocurriendo en paralelo. Cuando una de las funciones termina de procesar, toma el siguiente argumento y así sucesivamente. Eso es todo para el módulo de multiprocesamiento en Python. Referencia: Documentación oficial