Le pool multitraitement Python s'arrête brusquement

Question

J'essaie d'effectuer un traitement parallèle pour mes besoins, et le code semble fonctionner comme prévu pour les éléments 4k-5k en parallèle. Mais dès que les éléments à traiter commencent à augmenter, le code traite quelques listes puis sans générer d'erreur, le programme s'arrête brusquement.

J'ai vérifié et le programme n'est pas bloqué, la RAM est disponible (j'ai 16 Go de RAM) et l'utilisation du processeur n'est même pas de 30%. Je n'arrive pas à comprendre ce qui se passe. J'ai 1 million d'éléments à traiter.

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()

METTRE À JOUR -

Erreur trouvée - BrokenPipeError: [Errno 32] Pipe cassée

Trace -

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

Le code semble correct. La seule chose à laquelle je peux penser est que tous vos processus sont en attente d’achèvement. Voici une suggestion: au lieu d'utiliser le mécanisme de rappel fourni par apply_async, utilisez l' AsyncResultobjet qui est retourné pour obtenir la valeur de retour du processus. Tu peux appelergetsur cet objet en spécifiant une valeur de timeout (30 secondes arbitrairement spécifiées ci-dessous, peut-être pas assez longtemps). Si la tâche n'est pas terminée dans cette durée, une exception de délai d'expiration sera lancée (vous pouvez l'attraper, si vous le souhaitez). Mais cela testera l'hypothèse selon laquelle les processus sont suspendus. Assurez-vous simplement de spécifier une valeur de délai d'expiration suffisamment grande pour que la tâche se termine dans ce laps de temps. J'ai également divisé les soumissions de tâches en lots de 1 000, non pas parce que je pense que la taille de 1 000 000 est un problème en soi , mais simplement pour que vous n'ayez pas une liste de 1 000 000 objets de résultat. Mais si vous constatez que vous ne vous bloquez plus, essayez d'augmenter la taille du lot et voyez si cela fait une différence.

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()

Mettre à jour

Sur la base de Google sur plusieurs pages pour votre erreur de tuyau cassé, il semble que votre erreur puisse être le résultat d'une mémoire épuisée. Voir Multiprocessing Python: exception de tuyau cassé après l'augmentation de la taille du pool , par exemple. La refonte suivante tente d'utiliser moins de mémoire. Si cela fonctionne, vous pouvez alors essayer d'augmenter la taille du lot:

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()