Python падает, когда я добавляю process.join в скрипт при использовании многопроцессорной обработки

Question

Я изучал многопроцессорность и натолкнулся на пример этого на веб-сайте. Однако, когда я пытаюсь запустить этот пример на моей сетке MacBook, ничего не происходит. Следующим был пример:

import random 
import multiprocessing 


def list_append(count, id, out_list): 
""" 
Creates an empty list and then appends a 
random number to the list 'count' number 
of times. A CPU-heavy operation! 
""" 
for i in range(count): 
    out_list.append(random.random()) 

if __name__ == "__main__": 
size = 10000000 # Number of random numbers to add 
procs = 2 # Number of processes to create 

# Create a list of jobs and then iterate through 
# the number of processes appending each process to 
# the job list 
jobs = [] 
for i in range(0, procs): 
    out_list = list() 
    process = multiprocessing.Process(target=list_append, 
     args=(size, i, out_list)) 
    jobs.append(process) 

# Start the processes (i.e. calculate the random number lists)  
for j in jobs: 
    j.start() 

# Ensure all of the processes have finished 
for j in jobs: 
j.join() 

print ("List processing complete.") 

Как выясняется, после того как я положил заявление для печати в функции «list_append», ничего не печатается, так что проблема на самом деле не j.join(), а скорее j.start() немного.

Answer 1

При создании процесса с помощью multiprocessing.Process вы подготавливаете подфункцию, выполняемую асинхронно другим процессом. Вычисление начинается, когда вы вызываете метод start. Метод join ожидает, что вычисление будет выполнено. Поэтому, если вы только начинаете процесс и не ждете завершения (или join), ничего не произойдет, так как процесс будет убит, когда ваша программа выйдет.

Здесь одна проблема заключается в том, что вы не используете объект, разделяемый в multiprocessing. Когда вы используете общий list(), каждый процесс будет использовать другой список в памяти. Локальный процесс будет очищен при выходе процессов и в главном списке. Если вы хотите, чтобы иметь возможность обмениваться данными между процессами, вы должны использовать multiprocessing.Queue:

import random 
import multiprocessing 


def list_append(count, id, out_queue): 
    """ 
    Creates an empty list and then appends a 
    random number to the list 'count' number 
    of times. A CPU-heavy operation! 
    """ 
    for i in range(count): 
     out_queue.put((id, random.random())) 

if __name__ == "__main__": 
    size = 10000 # Number of random numbers to add 
    procs = 2 # Number of processes to create 

    # Create a list of jobs and then iterate through 
    # the number of processes appending each process to 
    # the job list 
    jobs = [] 
    q = multiprocessing.Queue() 
    for i in range(0, procs): 
     process = multiprocessing.Process(target=list_append, 
              args=(size, i, q)) 
     process.start() 
     jobs.append(process) 

    result = [] 
    for k in range(procs*size): 
     result += [q.get()] 

    # Wait for all the processes to finish 
    for j in jobs: 
     j.join() 

    print("List processing complete. {}".format(result)) 

Обратите внимание, что этот код может повиснуть довольно легко, если вы не вычислить правильно количество результатов, отправленных обратно в out_queue.
Если вы попытаетесь получить слишком много результатов, q.get будет ожидать дополнительного результата, который никогда не наступит. Если вы не получите весь результат от q, ваши процессы замерзнут, так как out_queue будет заполнен, а out_queue.put не вернется. Таким образом, ваши процессы никогда не выйдут, и вы не сможете присоединиться к ним.
Если ваши вычисления независимы, я настоятельно рекомендую посмотреть инструменты более высокого уровня, такие как Pool или даже более надежную стороннюю библиотеку, например joblib, так как она позаботится об этих аспектах для вас. (see this answer for some insights on Process vs Pool/joblib)

Я фактически уменьшил число size, так как программа станет медленной, если вы попытаетесь поместить многие объекты в Queue. Если вам нужно передать много мелких объектов, попробуйте передать их все в одном пакете:

import random 
import multiprocessing 


def list_append(count, id, out_queue): 
    a = [random.random() for i in range(count)] 
    out_queue.put((id, a)) 

if __name__ == "__main__": 
    size = 10000 # Number of random numbers to add 
    procs = 2 # Number of processes to create 
    jobs = [] 
    q = multiprocessing.Queue() 
    for i in range(0, procs): 
     process = multiprocessing.Process(target=list_append, 
              args=(size, i, q)) 
     process.start() 
     jobs.append(process) 
    result += [q.get() for _ in range(procs)] 
    for j in jobs: 
     j.join() 
    print("List processing complete.")