Multiproceso de Python: cómo limitar el número de procesos en espera?

Question

Cuando se ejecuta una gran cantidad de tareas (con parámetros grandes) usando Pool.apply_async, los procesos se asignan y pasan a un estado de espera, y no hay límite para la cantidad de procesos en espera. Esto puede terminar por el consumo de toda la memoria, como en el siguiente ejemplo:

import multiprocessing 
import numpy as np 

def f(a,b): 
    return np.linalg.solve(a,b) 

def test(): 

    p = multiprocessing.Pool() 
    for _ in range(1000): 
     p.apply_async(f, (np.random.rand(1000,1000),np.random.rand(1000))) 
    p.close() 
    p.join() 

if __name__ == '__main__': 
    test() 

Estoy buscando una manera de limitar la cola de espera, de tal manera que sólo hay un número limitado de procesos en espera, y Pool.apply_async está bloqueado mientras la cola de espera está llena.

Answer 1

multiprocessing.Pool tiene un _taskqueue miembro del tipo multiprocessing.Queue, que toma un parámetro opcional maxsize; lamentablemente lo construye sin el conjunto de parámetros maxsize.

me gustaría recomendar la subclasificación multiprocessing.Pool con un copy-paste de multiprocessing.Pool.__init__ que pasa maxsize-_taskqueue constructor.

mono-parchear el objeto (la piscina o la cola) también funcionaría, pero usted tendría que monkeypatch pool._taskqueue._maxsize y pool._taskqueue._sem por lo que sería muy frágil:

pool._taskqueue._maxsize = maxsize 
pool._taskqueue._sem = BoundedSemaphore(maxsize) 

Answer 2

Se podría añadir cola explícita con el parámetro maxsize y use queue.put() en lugar de pool.apply_async() en este caso. A continuación, los procesos de trabajo podrían:

for a, b in iter(queue.get, sentinel): 
    # process it 

Si desea limitar el número de argumentos de entrada creadas/resultados que se encuentran en la memoria a aproximadamente el número de procesos de trabajo activas Posteriormente, se podría utilizar pool.imap*() métodos:

#!/usr/bin/env python 
import multiprocessing 
import numpy as np 

def f(a_b): 
    return np.linalg.solve(*a_b) 

def main(): 
    args = ((np.random.rand(1000,1000), np.random.rand(1000)) 
      for _ in range(1000)) 
    p = multiprocessing.Pool() 
    for result in p.imap_unordered(f, args, chunksize=1): 
     pass 
    p.close() 
    p.join() 

if __name__ == '__main__': 
    main() 

Answer 3

esperar si pool._taskqueue es sobre el tamaño deseado:

import multiprocessing 
import numpy as np 
import time 

def f(a,b): 
    return np.linalg.solve(a,b) 

def test(max_apply_size=100): 
    p = multiprocessing.Pool() 
    for _ in range(1000): 
     p.apply_async(f, (np.random.rand(1000,1000),np.random.rand(1000))) 

     while pool._taskqueue.qsize() > max_apply_size: 
      time.sleep(1) 

    p.close() 
    p.join() 

if __name__ == '__main__': 
    test()