Использование многопроцессорной блокировки () для предотвращения создания поврежденных графиков

Question

Когда для бэкэнда matplotlib установлено значение «PDF», я включил построение графиков в функцию. Центральная тема функции заключается в создании графиков, например:

def func(arg1,arg2,arg3,num):
    try:
        fig = pdf.savefig(num+arg1+arg2,bbox_inches = "tight")
        return fig
    except:
        return None

Мне удалось успешно получить желаемые results (графики) последовательным способом со следующей реализацией:

data = list(range(100))
results=[func(arg1,arg2,arg3,num) for num in data]

Я попытался реализовать это параллельно, используя методы pool.map () и pool.apply_async () следующим образом:

Реализация pool.map ():

if __name__ == "__main__":
    try:
        pool = Pool()    
        data = list(range(50))
        funct = partial(func,arg1,arg2,arg3)
        results = pool.map(funct, data)
    finally:
        pool.close()
        pool.join()

Реализация pool.async ():

if __name__ == "__main__":
    try:
        pool = Pool()    
        results = []
        data = list(range(50))
        result_objects = [pool.apply_async(func, args=(arg1,arg2,arg3,num)) for num in data]
        results = [r.get() for r in result_objects]
    finally:
        pool.close()
        pool.join()

В обеих параллельных реализациях я заметил, что из 50 симуляций только шесть приводят к читаемому PDF-файлу, а другие повреждены. Когда я изменяю число симуляций, скажем, десять, только три приводят к читаемому PDF-файлу, а другие повреждены.

Кажется, я не понимаю, почему только несколько сюжетов создаются должным образом, а другие повреждены.

Я выполняю многопроцессорную обработку на 4-ядерном компьютере с Linux Ubuntu 18.04.

Я столкнулся с многопроцессорным модулем queue(), который, кажется, заботится о связи между главным и дочерними процессами. Я предполагаю, что есть какая-то ошибка в том, как происходит общение в настоящее время, что приводит к искажению изображений для большинства итераций.

Требуется помощь во включении многопроцессорной обработки queue() в код для решения этой проблемы.

Answer 1

Я предполагал, что Queue () может обладать магическими способностями сгладить морщины в коде.

Просматривая несколько сообщений , я понял, что у Lock () есть потенциал, чтобы разблокировать мои страдания.

Следующий модифицированный код работает нормально:

Реализация pool.map ():

def func(arg1,arg2,arg3,lock,num):
    with lock:
        #your code#

from multiprocessing import Pool, Lock, Manager
from functools import partial
import time

if __name__ == "__main__":
    try: 
        pool = Pool()
        m = Manager()
        lock = m.Lock()
        data = list(range(1000))
        funct = partial(func,arg1,arg2,arg3,lock)
        results = pool.map(funct, data)
    except:
        pool.close()
        pool.join()

Реализация pool.async ():

def func(num,arg1,arg2,arg3,lock):
    with lock:
        # your code #

from multiprocessing import Pool, Lock, Manager
from functools import partial
import time

if __name__ == "__main__":
    try:
        pool = Pool()
        m = Manager()
        lock = m.Lock()
        results = []
        data = list(range(1000))
        result_objects = [pool.apply_async(func, args=(num,arg1,arg2,arg3,lock)) for num in data]
        results = [r.get() for r in result_objects]
    finally:
        pool.close()
        pool.join()