Python Çoklu İşlem Havuzu Aniden Duruyor

Question

Gereksinimlerim için paralel işlem yapmaya çalışıyorum ve kod 4k-5k elemanlar için paralel olarak beklendiği gibi çalışıyor gibi görünüyor. Ancak işlenecek elemanlar artmaya başlar başlamaz, kod birkaç listeyi işler ve sonra herhangi bir hata atmadan program aniden çalışmayı durdurur.

Kontrol ettim ve program askıda değil, RAM mevcut (16 Gb RAM'im var) ve CPU Kullanımı% 30 bile değil. Ne olduğunu anlayamıyorum. İşlenecek 1 milyon elementim var.

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

GÜNCELLEME -

Hata bulundu - BrokenPipeError: [Errno 32] Kırık boru

İz -

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

Kod doğru görünüyor. Aklıma gelen tek şey, tüm süreçlerinizin bitmeyi beklemekte olmasıdır. İşte bir öneri: tarafından sağlanan geri arama mekanizmasını kullanmak yerine , işlemden dönüş değerini almak için döndürülen nesneyi apply_asynckullanın AsyncResult. Arayabilirsingetbir zaman aşımı değeri belirten bu nesnede (aşağıda keyfi olarak 30 saniye belirtilmiş, muhtemelen yeterince uzun değil). Görev bu süre içinde tamamlanmadıysa, bir zaman aşımı istisnası atılır (isterseniz onu yakalayabilirsiniz). Ancak bu, süreçlerin asılı olduğu hipotezini test edecek. Görevin o süre içinde tamamlaması için yeterince büyük bir zaman aşımı değeri belirttiğinizden emin olun. Ayrıca görev gönderimlerini 1000'lik gruplara ayırdım, 1.000.000 büyüklüğünün başlı başına bir sorun olduğunu düşündüğümden değil, 1.000.000 sonuç nesnesi listeniz olmadığı için. Ancak sonuç olarak artık takılmadığınızı fark ederseniz, parti boyutunu artırmayı deneyin ve bunun bir fark yaratıp yaratmadığını görün.

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

Güncelleme

Bozuk boru hatanız için Google'da birkaç sayfadan yararlanmaya dayalı olarak, hatanızın bellek tüketiminin bir sonucu olabileceği görülüyor. Örneğin, Havuz boyutunu artırdıktan sonra Python Çoklu İşlem: Kırık Boru istisnasına bakın . Aşağıdaki yeniden çalışma , daha az bellek kullanmaya çalışır . Çalışırsa, parti boyutunu artırmayı deneyebilirsiniz:

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