Python: programación multiproceso en acción

Vamos directos a ver cómo crear un proceso con el módulo multiprocessing.Process. El código está comentado para que entiendas que se hace en cada parte:

from multiprocessing import Process


def worker(num):
    print(f"Worker: {num}")


if __name__ == "__main__":
    # 1. Crea el objeto de tipo Process
    process = Process(target=worker, args=(1,))

    # 2. Lanza el proceso
    process.start()

    # 3. Se espera a que termine el proceso
    process.join()

Este programa crear un proceso que ejecuta la función worker. El proceso padre crea el proceso con Process, lo lanza/ejecuta llamando a start() y espera que termine con join(). Esta es la estructura básica.

La clase Process necesite que indiques como target la función que se va a ejecutar cuando se llame a start(). Si la función que indicas en target necesita argumentos, estos se indican por medio de args que recibe la tupla con la lista de valores a pasar a la lista de argumentos.

Primero el código del programa en Python:

from multiprocessing import Process, current_process
from time import sleep
from random import randint


def task(n1, n2):
    p = current_process()
    sleep(randint(1, 3))
    print(f"Suma proceso {p.pid}: {n1} + {n2} = {n1 + n2}")


if __name__ == "__main__":
    for _ in range(5):
        process = Process(target=task, args=(10, 15))
        process.start()

Este programa (proceso padre) crea 5 procesos hijos que realizan la misma suma.

• Process(target=task, args=(10, 15)): la clase Process es la que usamos para crear un proceso nuevo (subproceso o proceso hijo del actual). Necesita que le indiquemos la función que se ejecutará en el nuevo proceso hijo, a través del argumento target, y, si dicha función necesita argumentos, hay que pasarle los argumentos por medio de una tupla en args.
• process.start(): tenemos que llamar al método start para que el proceso hijo, ya creado, comience su ejecución.
• current_process(): con esta función podemos obtener el proceso que está ejecutando el código desde el que se llama.

En el código anterior se están lanzando los procesos hijos sin ningún control sobre los mismos por parte del padre. Hay situaciones y escenarios donde vamos a necesitar sincronizar padre e hijos.

En el siguiente ejemplo tenemos una versión del programa anterior en el que el padre espera que terminen todos los hijos antes de acabar él mismo para imprimir un mensaje final por pantalla.

Se usa el método join que bloquea al padre hasta que el hijo concreto termine:

from multiprocessing import Process, current_process
from time import sleep
from random import randint


def task(n1, n2):
    p = current_process()
    sleep(randint(1, 3))
    print(f"Suma proceso {p.pid}: {n1} + {n2} = {n1 + n2}")


if __name__ == "__main__":
    processes = [ Process(target=task, args=(10, 15)) for _ in range(5) ]

    for p in processes:
        p.start()

    for p in processes:
        p.join()

    print(f"Proceso padre, con PID={current_process().pid} terminado")

Primero se crean los procesos, luego se lanzan para que se ejecuten en paralelo y, por último, el proceso padre espera a que todos terminen.

for p in processes:
    p.start()
    p.join()

El módulo multiprocessing nos ofrece gran cantidad de propiedades y funciones con las que podemos obtener información relativa a los procesos.

Este es un ejemplo en el que se usan las funciones:

• current_process para obtener el proceso que está ejecutando la función o el código donde se llama.
• parent_process para obtener el proceso padre que creó el proceso hijo desde el que se llama a esta función.
• cpu_count para obtener el número de cores que tiene la CPU en la que se ejecuta el programa desde el cual se llama a dicha función.

Aquí puedes ver un ejemplo en el que se usan estas dos funciones para mostrar información adicional para completar el programa anterior:

from multiprocessing import Process, current_process, parent_process, cpu_count
from time import sleep
from random import randint


def task(n1, n2):
    p = current_process()
    sleep(randint(1, 3))
    print(f"Suma proceso {p.pid}: {n1} + {n2} = {n1 + n2}  |  Proceso padre: {parent_process().pid}")


if __name__ == "__main__":
    print(f"Ejecutado con CPU con un total de {cpu_count()} cores")

    processes = [ Process(target=task, args=(10, 15)) for _ in range(5) ]

    for p in processes:
        p.start()

    for p in processes:
        p.join()

    print(f"Proceso padre, con PID={current_process().pid} terminado")

En escenarios más complejos, donde la lógica de los procesos hijos se complica, heredar de la clase Process puede ser una buena solución.

Crear clases que heredan de la clase Process te permitirá, además, tener un código mejor organizado, más fácil de probar y mantener.

Para crar una subclase de Process solo tienes que sobreescribir el método run que es el método que se ejecuta, de la clase Process, cuando se llama al método start.

Si necesitas pasar datos de entrada al proceso hijo, hazlo a través del constructor para disponer de esos datos vía atributos de clase. Aquí, el ejemplo en el que se crean varios procesos que suman usando una clase derivada en vez de una función:

from multiprocessing import Process, current_process
from time import sleep
from random import randint


class Adder(Process):

    def __init__(self, n1, n2):
        super().__init__()

        self._n1 = n1
        self._n2 = n2

    def run(self):
        sleep(randint(1, 3))
        print(f"Suma proceso con PID {current_process().pid}: {self._n1} + {self._n2} = {self._n1 + self._n2}")


if __name__ == "__main__":
    processes = [ Adder(10, 15) for _ in range(5) ]

    for p in processes:
        p.start()

    for p in processes:
        p.join()

    print(f"Proceso padre, con PID={current_process().pid} terminado")

Vamos a modificar el programa anterior:

• En primer lugar, cada proceso hijo va a realizar la suma de dos números aleatorios (hasta ahora siempre sumaban el número 10 y el número 15).
• Y, en segundo lugar, queremos realizar el promedio de las sumas de los 5 procesos hijos. Esta media la hará el padre.

Estos cambios nos obligan a compartir datos entre los hijos y el padre, ya que este último necesita conocer el resultado de las sumas para hacer el promedio.

Para terminar de entender lo que queremos hacer, veamos un caso de uso:

• El proceso 1 suma los número 3 y 7: 3 + 7 = 10
• El proceso 2 suma los número 1 y 3: 1 + 3 = 4
• El proceso 3 suma los número 4 y 4: 4 + 4 = 8
• El proceso 4 suma los número 9 y 5: 9 + 5 = 14
• El proceso 5 suma los número 6 y 3: 6 + 3 = 9
• De alguna manera vamos a tener que sumar los resultados: 10 + 4 + 8 + 14 + 9 = 45
• Y, por último, obtener el promedio: 45 / 5 = 9

Una solución podría ser usar una variable global que usen todos los procesos hijos para añadir sus resultados. Vamos directos al código:

from multiprocessing import Process, current_process
from time import sleep
from random import randint

# Variable global para acumular las sumas
shared_total: int = 0


class Adder(Process):

    def __init__(self, n1, n2):
        super().__init__()

        self._n1 = n1
        self._n2 = n2

    def run(self):
        # En Python tenemos que usar la instrucción global para indicar que
        # la variable que vamos a usar aquí es la variable global.
        # Si no lo hacemos así va a crear una variable local a este método
        # llamada shared_local.
        global shared_total

        sleep(randint(1, 3))
        shared_total += self._n1 + self._n2
        print(f"Suma proceso con PID {current_process().pid}: {self._n1} + {self._n2} = {self._n1 + self._n2}")
        print(f"    Acumulado hasta ahora: {shared_total}")
        print()


if __name__ == "__main__":
    processes = [ Adder(randint(1, 10), randint(1, 10)) for _ in range(5) ]

    for p in processes:
        p.start()

    for p in processes:
        p.join()

    print(f"Proceso padre, con PID={current_process().pid}")
    print(f"    Promedio: {shared_total} / 5 = {shared_total / 5}")

Al ejecutar el programa, vemos en la salida "cosas extrañas":

Suma proceso con PID 28931: 2 + 4 = 6
    Acumulado hasta ahora: 6

Suma proceso con PID 28932: 1 + 6 = 7
    Acumulado hasta ahora: 7

Suma proceso con PID 28933: 8 + 10 = 18
    Acumulado hasta ahora: 18

Suma proceso con PID 28934: 8 + 9 = 17
    Acumulado hasta ahora: 17

Suma proceso con PID 28935: 10 + 7 = 17
    Acumulado hasta ahora: 17

Proceso padre, con PID=28930
    Promedio: 0 / 5 = 0.0

• El valor de la variable global shared_total en cada proceso coincide con la suma parcial que hacen (no acumula como esperábamos).
• El proceso padre se encuentra que la variable global shared_total es igual a 0 (el valor con el que se inicializó dicha variable).

No hay nada extraño, en realidad, si lo pensamos bien y recordamos la parte teórica de procesos en la que ya hablamos de que cada proceso tiene su propio espacio en memoria. Esto significa, en la práctica, y en este ejemplo, que cada proceso tiene su propia copia de shared_total.

Existen varios métodos de comunicación entre procesos. Nosotros nos vamos a centrar en dos:

• Memoria Compartida: este método permite que varios procesos accedan a la misma región de memoria. Es muy eficiente porque evita la necesidad de copiar datos entre procesos. Sin embargo, requiere mecanismos de sincronización (como semáforos o mutexes) para evitar condiciones de carrera (lo veremos más adelante).
• Ficheros: los procesos pueden leer y escribir en ficheros en el sistema de archivos. Este método es más sencillo de implementar y no requiere sincronización en la memoria, pero puede ser más lento debido a la necesidad de acceder al disco.

Se llama sección o región crítica, en programación concurrente, a la porción de código de un programa en la que se accede a un recurso compartido que no debe ser accedido por más de un proceso o hilo.

Para manejar estas secciones críticas existen diferentes mecanismos que permiten sincronizar a estos procesos y/o hilos que compiten por el recurso compartido. Nosotros vamos a estudiar dos:

• Locks que permiten restringir el acceso a un recurso compartido para que solo un proceso/hilo acceda a él y no se produzcan accesos simultáneaos.
• Semáforos que son un caso concreto de lock en el que se permite el acceso a una sección crítica a uno o varios procesos.

Cuando usamos locks o semáforos, si un proceso intenta acceder al recurso compartido bajo dicho lock o semáforo, entonces el proceso/hilo quedará bloqueado hasta que se libere.