El grupo de multiprocesamiento de Python se detiene abruptamente

Question

Estoy tratando de realizar un procesamiento paralelo para mis requisitos, y el código parece funcionar como se esperaba para elementos 4k-5k en paralelo. Pero tan pronto como los elementos a procesar comienzan a aumentar, el código procesa algunos listados y luego, sin arrojar ningún error, el programa deja de ejecutarse abruptamente.

Lo verifiqué y el programa no se cuelga, la RAM está disponible (tengo una RAM de 16 Gb) y la utilización de la CPU ni siquiera es del 30%. Parece que no puedo entender qué está pasando. Tengo 1 millón de elementos para procesar.

def get_items_to_download():
    #iterator to fetch all items that are to be downloaded
    yield download_item

def start_download_process():
    multiproc_pool = multiprocessing.Pool(processes=10)
    for download_item in get_items_to_download():
        multiproc_pool.apply_async(start_processing, args = (download_item, ), callback = results_callback)
    
    multiproc_pool.close()
    multiproc_pool.join()

def start_processing(download_item):
    try:
        # Code to download item from web API
        # Code to perform some processing on the data
        # Code to update data into database
        return True
    except Exception as e:
        return False

def results_callback(result):
    print(result)

if __name__ == "__main__":
    start_download_process()

ACTUALIZAR -

Encontré el error: BrokenPipeError: [Errno 32] Tubería rota

Traza -

Traceback (most recent call last):
File "/usr/lib/python3.6/multiprocessing/pool.py", line 125, in worker
put((job, i, result))
File "/usr/lib/python3.6/multiprocessing/queues.py", line 347, in put
self._writer.send_bytes(obj)
File "/usr/lib/python3.6/multiprocessing/connection.py", line 200, in send_bytes
self._send_bytes(m[offset:offset + size])
File "/usr/lib/python3.6/multiprocessing/connection.py", line 404, in _send_bytes
self._send(header + buf)
File "/usr/lib/python3.6/multiprocessing/connection.py", line 368, in _send
n = write(self._handle, buf)
BrokenPipeError: [Errno 32] Broken pipe

Answer 1

def get_items_to_download():
    #instead of yield, return the complete generator object to avoid iterating over this function.
    #Return type - generator (download_item1, download_item2...)
    return download_item


def start_download_process():
    download_item = get_items_to_download()
    # specify the chunksize to get faster results. 
    with multiprocessing.Pool(processes=10) as pool:
    #map_async() is also available, if that's your use case.
        results= pool.map(start_processing, download_item, chunksize=XX )  
    print(results)
    return(results)

def start_processing(download_item):
    try:
        # Code to download item from web API
        # Code to perform some processing on the data
        # Code to update data into database
        return True
    except Exception as e:
        return False

def results_callback(result):
    print(result)

if __name__ == "__main__":
    start_download_process()

Answer 2

El código parece correcto. Lo único en lo que puedo pensar es en que todos sus procesos están pendientes esperando su finalización. Aquí hay una sugerencia: en lugar de usar el mecanismo de devolución de llamada proporcionado por apply_async, use el AsyncResultobjeto que se devuelve para obtener el valor de retorno del proceso. Puedes llamargeten este objeto especificando un valor de tiempo de espera (30 segundos especificados arbitrariamente a continuación, posiblemente no lo suficiente). Si la tarea no se ha completado en esa duración, se lanzará una excepción de tiempo de espera (puede detectarla, si lo desea). Pero esto pondrá a prueba la hipótesis de que los procesos están suspendidos. Solo asegúrese de especificar un valor de tiempo de espera que sea lo suficientemente grande como para que la tarea se complete dentro de ese período de tiempo. También he dividido los envíos de tareas en lotes de 1000, no porque crea que el tamaño de 1,000,000 sea un problema per se , sino para que no tenga una lista de 1,000,000 de objetos de resultado. Pero si descubre que ya no se cuelga como resultado, intente aumentar el tamaño del lote y vea si hace una diferencia.

import multiprocessing

def get_items_to_download():
    #iterator to fetch all items that are to be downloaded
    yield download_item

BATCH_SIZE = 1000

def start_download_process():
    with multiprocessing.Pool(processes=10) as multiproc_pool:
        results = []
        for download_item in get_items_to_download():
            results.append(multiproc_pool.apply_async(start_processing, args = (download_item, )))
            if len(results) == BATCH_SIZE:
                process_results(results)
                results = []
        if len(results):
            process_results(results)
    

def start_processing(download_item):
    try:
        # Code to download item from web API
        # Code to perform some processing on the data
        # Code to update data into database
        return True
    except Exception as e:
        return False

TIMEOUT_VALUE = 30 # or some suitable value

def process_results(results):
    for result in results:
        return_value = result.get(TIMEOUT_VALUE) # will cause an exception if process is hanging
        print(return_value)

if __name__ == "__main__":
    start_download_process()

Actualizar

Basado en Google varias páginas para su error de tubería rota, parece que su error podría ser el resultado de agotar la memoria. Consulte Multiprocesamiento de Python: Excepción de tubería rota después de aumentar el tamaño del grupo , por ejemplo. La siguiente reelaboración intenta utilizar menos memoria. Si funciona, puede intentar aumentar el tamaño del lote:

import multiprocessing


BATCH_SIZE = 1000
POOL_SIZE = 10


def get_items_to_download():
    #iterator to fetch all items that are to be downloaded
    yield download_item


def start_download_process():
    with multiprocessing.Pool(processes=POOL_SIZE) as multiproc_pool:
        items = []
        for download_item in get_items_to_download():
            items.append(download_item)
            if len(items) == BATCH_SIZE:
                process_items(multiproc_pool, items)
                items = []
        if len(items):
            process_items(multiproc_pool, items)


def start_processing(download_item):
    try:
        # Code to download item from web API
        # Code to perform some processing on the data
        # Code to update data into database
        return True
    except Exception as e:
        return False


def compute_chunksize(iterable_size):
    if iterable_size == 0:
        return 0
    chunksize, extra = divmod(iterable_size, POOL_SIZE * 4)
    if extra:
        chunksize += 1
    return chunksize


def process_items(multiproc_pool, items):
    chunksize = compute_chunksize(len(items))
    # you must iterate the iterable returned:
    for return_value in multiproc_pool.imap(start_processing, items, chunksize):
        print(return_value)


if __name__ == "__main__":
    start_download_process()