PNL | Procesamiento de listas en paralelo con execnet

Este artículo presenta un patrón para usar execnet para procesar una lista en paralelo. Es un patrón de función para mapear cada elemento de la lista a un nuevo valor, usando execnet para hacer el mapeo en paralelo.

En el código que se proporciona a continuación, los números enteros simplemente se duplican, se puede realizar cualquier cálculo puro. Dado es el módulo, que será ejecutado por execnet. Recibe una tupla de 2 de (i, arg), asume que arg es un número y devuelve (i, arg*2).

Código:

if __name__ == '__channelexec__':
    for (i, arg) in channel:
        channel.send((i, arg * 2))

Para usar este módulo para duplicar todos los elementos de una lista, importe el módulo plists y llame a plists.map() con el módulo remote_double y una lista de enteros para duplicar.

Código: usando plist

import plists, remote_double
plists.map(remote_double, range(10))

Producción :

[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

La función map() se define en plists.py. Toma un módulo puro, una lista de argumentos y una lista opcional de 2 tuplas que consisten en (especificación, conteo). Las especificaciones predeterminadas son [(‘popen’, 2)], lo que significa que el usuario abrirá dos puertas de enlace y canales locales. Una vez que se abren estos canales, el usuario puede ponerlos en un ciclo de itertools, lo que crea un iterador infinito que vuelve al principio una vez que llega al final.

Ahora, cada argumento se puede enviar en args a un canal para su procesamiento, y dado que los canales se ciclan, cada canal obtiene una distribución de argumentos casi uniforme. Aquí es donde entra i : se desconoce el orden en que se obtienen los resultados, por lo que i , como índice de cada argumento de la lista, se pasa al canal y viceversa para que el usuario pueda combinar los resultados en el orden original. Luego, espere los resultados con una cola de recepción multicanal e insértelos en una lista precargada que tenga la misma longitud que los argumentos originales. Después de obtener todos los resultados esperados, salga de las puertas de enlace y devuelva los resultados como se muestra en el código que figura a continuación:

Código:

import itertools, execnet
def map(mod, args, specs =[('popen', 2)]):
    gateways = []
    channels = []
      
    for spec, count in specs:
        for i in range(count):
            gw = execnet.makegateway(spec)
            gateways.append(gw)
            channels.append(gw.remote_exec(mod))
              
    cyc = itertools.cycle(channels)
      
    for i, arg in enumerate(args):
        channel = next(cyc)
        channel.send((i, arg))
    mch = execnet.MultiChannel(channels)
    queue = mch.make_receive_queue()
    l = len(args)
    # creates a list of length l, 
    # where every element is None
    results = [None] * l 
      
    for i in range(l):
        channel, (i, result) = queue.get()
        results[i] = result
          
    for gw in gateways:
        gw.exit()
    return results

Código: aumentar la paralelización modificando las especificaciones

plists.map(remote_double, range(10), [('popen', 4)])

Producción :

[0, 2, 4, 6, 8, 10, 12, 14, 16, 18]

Sin embargo, una mayor paralelización no significa necesariamente un procesamiento más rápido. Depende de los recursos disponibles, y cuantas más puertas de enlace y canales se abran, más gastos generales se requieren. Idealmente, debería haber una puerta de enlace y un canal por núcleo de CPU para obtener la máxima utilización de recursos. Use plists.map() con cualquier módulo puro siempre que reciba y envíe 2 tuplas donde i es el primer elemento. Este patrón es más útil cuando hay un montón de números para procesar lo más rápido posible.