Python Multiprocessing Pool หยุดกะทันหัน

Question

ฉันกำลังพยายามประมวลผลแบบขนานสำหรับความต้องการของฉันและดูเหมือนว่าโค้ดจะทำงานตามที่คาดไว้สำหรับองค์ประกอบ 4k-5k ในแบบขนาน แต่ทันทีที่องค์ประกอบที่ต้องประมวลผลเริ่มเพิ่มขึ้นโค้ดจะประมวลผลรายชื่อสองสามรายการจากนั้นโดยไม่มีข้อผิดพลาดใด ๆ โปรแกรมจะหยุดทำงานทันที

ฉันตรวจสอบแล้วโปรแกรมไม่ได้หยุดทำงานมี RAM (ฉันมีแรม 16 Gb) และการใช้งาน CPU ไม่ถึง 30% ดูเหมือนจะคิดไม่ออกว่าเกิดอะไรขึ้น ฉันมี 1 ล้านองค์ประกอบที่ต้องดำเนินการ

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

อัปเดต -

พบข้อผิดพลาด - BrokenPipeError: [Errno 32] ท่อแตก

ติดตาม -

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

รหัสถูกต้อง สิ่งเดียวที่ฉันคิดได้คือกระบวนการทั้งหมดของคุณกำลังรอให้เสร็จสิ้น นี่คือคำแนะนำ: แทนที่จะใช้กลไกการโทรกลับที่มีให้ให้apply_asyncใช้AsyncResultอ็อบเจกต์ที่ส่งคืนเพื่อรับค่าส่งคืนจากกระบวนการ คุณสามารถโทรgetบนวัตถุนี้ระบุค่าการหมดเวลา (30 วินาทีที่ระบุโดยพลการด้านล่างอาจไม่นานพอ) หากงานยังไม่เสร็จสิ้นในช่วงเวลาดังกล่าวข้อยกเว้นการหมดเวลาจะถูกโยนทิ้ง (คุณสามารถจับได้หากต้องการ) แต่นี่จะเป็นการทดสอบสมมติฐานที่ว่ากระบวนการต่างๆกำลังแขวนอยู่ เพียงตรวจสอบให้แน่ใจว่าได้ระบุค่าการหมดเวลาที่มากพอที่งานควรจะเสร็จสิ้นภายในช่วงเวลานั้น ฉันยังแยกการส่งงานออกเป็นชุดละ 1,000 ไม่ใช่เพราะฉันคิดว่าขนาดของ 1,000,000 เป็นปัญหาต่อหนึ่งแต่คุณไม่มีรายการวัตถุผลลัพธ์ 1,000,000 รายการ แต่ถ้าคุณพบว่าคุณไม่ค้างอีกต่อไปแล้วให้ลองเพิ่มขนาดแบทช์และดูว่ามันสร้างความแตกต่างหรือไม่

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

อัปเดต

จาก Googling หลายหน้าสำหรับข้อผิดพลาดท่อแตกของคุณดูเหมือนว่าข้อผิดพลาดของคุณอาจเป็นผลมาจากหน่วยความจำที่ใช้งานจนหมด ดูPython Multiprocessing: Broken Pipe ข้อยกเว้นหลังจากเพิ่มขนาด Poolเป็นต้น การทำซ้ำต่อไปนี้จะพยายามใช้หน่วยความจำน้อยลง หากได้ผลคุณสามารถลองเพิ่มขนาดแบทช์:

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