Python Multiprocessing Pool stoppt abrupt

Question

Ich versuche, eine parallele Verarbeitung für meine Anforderungen durchzuführen, und der Code scheint für 4k-5k-Elemente parallel wie erwartet zu funktionieren. Sobald jedoch die zu verarbeitenden Elemente zunehmen, verarbeitet der Code einige Einträge, und ohne Fehler wird das Programm abrupt nicht mehr ausgeführt.

Ich habe überprüft und das Programm ist nicht hängen geblieben, der RAM ist verfügbar (ich habe einen 16 Gb RAM) und die CPU-Auslastung beträgt nicht einmal 30%. Ich kann nicht herausfinden, was passiert. Ich habe 1 Million Elemente zu verarbeiten.

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

UPDATE -

Fehler gefunden - BrokenPipeError: [Errno 32] Pipe gebrochen

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

Der Code sieht korrekt aus. Das einzige, woran ich denken kann, ist, dass alle Ihre Prozesse auf den Abschluss warten. Hier ein Vorschlag: Verwenden Sie anstelle des von bereitgestellten Rückrufmechanismus apply_asyncdas zurückgegebene AsyncResultObjekt, um den Rückgabewert aus dem Prozess abzurufen. Du kannst anrufengetfür dieses Objekt einen Zeitlimitwert angeben (30 Sekunden, die unten willkürlich angegeben werden, möglicherweise nicht lang genug). Wenn die Aufgabe in dieser Zeit nicht abgeschlossen wurde, wird eine Timeout-Ausnahme ausgelöst (Sie können sie abfangen, wenn Sie dies wünschen). Dies wird jedoch die Hypothese überprüfen, dass die Prozesse hängen. Stellen Sie nur sicher, dass Sie einen Zeitlimitwert angeben, der ausreichend groß ist, damit die Aufgabe innerhalb dieses Zeitraums abgeschlossen werden kann. Ich habe auch die Aufgabenübermittlungen in Stapel von 1000 aufgeteilt, nicht weil ich denke, dass die Größe von 1.000.000 per se ein Problem darstellt , sondern nur, damit Sie keine Liste mit 1.000.000 Ergebnisobjekten haben. Wenn Sie jedoch feststellen, dass Sie dadurch nicht mehr hängen bleiben, versuchen Sie, die Stapelgröße zu erhöhen, und prüfen Sie, ob dies einen Unterschied macht.

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

Aktualisieren

Basierend auf dem Googeln mehrerer Seiten für Ihren Rohrbruchfehler scheint es, dass Ihr Fehler das Ergebnis einer Erschöpfung des Speichers sein könnte. Siehe Python-Multiprocessing: Broken Pipe-Ausnahme, nachdem Sie beispielsweise die Poolgröße erhöht haben . Die folgende Überarbeitung versucht , weniger Speicher zu verwenden. Wenn es funktioniert, können Sie versuchen, die Stapelgröße zu erhöhen:

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