Почему рабочие умирают от проблем с памятью при использовании многопроцессорной обработки в Python для вычисления LCA?

Question

У меня возникла проблема с использованием многопроцессорной обработки в Python для вычисления результатов LCA для всех функциональных блоков в базе данных ecoinvent v3.2 для нескольких итераций.

Код следующий:

for worker_id in range(CPUS):

    # Create child processes that can work apart from parent process
    child = mp.Process(target=worker_process, args=(projects.current, output_dir, worker_id, activities, ITERATIONS, status))
    workers.append(child)
    child.start()
print(workers)

while any(i.is_alive() for i in workers):
    time.sleep(0.1)
    while not status.empty():
        # Flush queue of progress reports
        worker, completed = status.get()
        progress[worker] = completed
    progbar.update(sum(progress.values()))
progbar.finish()

Определение worker_process функционировать следующим образом:

def worker_process(project, output_dir, worker_id, activities, iterations, progress_queue):

# Project is string; project name in Brightway2
# output_dir is a string
# worker_id is an integer
# activities is a list of dictionaries
# iterations is an integer
# progress_queue is a Queue where we can report progress to parent process

projects.set_current(project, writable=False)

lca = DirectSolvingPVLCA(activities[0])
lca.load_data()
samples = np.empty((iterations, lca.params.shape[0]))
supply_arrays = np.empty((iterations, len(activities), len(lca.product_dict)))

for index in range(iterations):

    lca.rebuild_all()
    samples[index, :] = lca.sample
    lca.decompose_technosphere()  
    for act_index, fu in enumerate(activities):
        lca.build_demand_array(fu)
        supply_arrays[index, act_index, :] = lca.solve_linear_system()
    progress_queue.put((worker_id, index))

Наблюдаемые проблемы:

1. Для более чем двух работников все, кроме двух, умирают немедленно от MemoryError (увидеть ниже).

2. Для двух выживших работников, кажется, код работает нормально для 10, 100 или 5000 функциональных блоков, но, как мы просим все FU, он ломается и сталкивается с тем же MemoryError,

это MemoryError происходит для каждого процесса X:

    Process Process-X: 
    Traceback (most recent call last): 
      File "C:\bw2-python\envs\bw2\lib\multiprocessing\process.py", line 254, in_bootstrap 
        self.run()    
      File "C:\bw2-python\envs\bw2\lib\multiprocessing\process.py", line 93, in run   
        self._target(*self._args, **self._kwargs)   
      File "C:\test\Do all the calculations.py", line 49, in worker_process 
        supply_arrays = np.empty((iterations, len(activities), len(lca.product_dict))) 
    MemoryError

Мои вопросы:

• Почему это происходит?

• Как это можно исправить?

Answer 1

Вам не хватает памяти, потому что вы используете слишком много памяти.

Когда вы выделяете новый массив с помощью:

np.empty((iterations, len(activities), len(lca.product_dict))) 

А также activities а также lca.product_dict каждый из них имеет длину, скажем, 10.000, вы используете 10.000 * 10.000 * 8 (при условии, что значение по умолчанию равно 64 битам или 8 байтам) = 800 МБ оперативной памяти на одну итерацию и на рабочий процесс.

Простым решением является работа на сервере с большим количеством оперативной памяти.

Альтернативы созданию этих больших массивов в памяти включают в себя:

• Использование PyTables или h5py
• Используя Numpy'smemmap,

В любом случае вам нужно будет тщательно протестировать наиболее эффективные способы записи и чтения данных для вашего конкретного рабочего процесса и ОС.