Как реализовать многопроцессорность в Python?

Question

Я хочу использовать многопроцессорность в Python для сортировки независимых списков. Например, у меня есть словарь int в качестве ключа и список в качестве значения.

Я пытался реализовать простую программу, но мне трудно снова сохранить отсортированный список в defaultdict и вернуть его в основной модуль.

from multiprocessing import Process

def fun(id, user_data):
    user_data.sort()
    return user_data

# users_data is a defaultdict of id as key and a list as a value
 
if __name__ == '__main__':
    for id,user_data in users_data.items():
        P= Process(target=fun,args=(id,user_data))
        P.start()
        P.join()    

Answer 1

Вам нужно будет использовать Manager для обмена данными между процессами. Кроме того, как @Tomerikoo упоминал в комментариях, то, как вы это делаете прямо сейчас, на самом деле не приведет к многопроцессорной обработке, поскольку P.join() сразу после P.start() приостановит скрипт, чтобы этот процесс завершился sh, что приведет к последовательное выполнение, а не параллельное.

Вы можете сделать что-то вроде этого:

from multiprocessing import Process, Manager

def sort_list(user_id, user_data, interprocess_dict):
    user_data.sort()
    interprocess_dict[user_id] = user_data


users_data = {}
users_data[1] = [5, 2, 1]
users_data[3] = [10, 12, 1]


def main():
    interprocess_dict = Manager().dict()
    processes = []
    for user_id, user_data in users_data.items():
        proc = Process(target=sort_list, args=(user_id, user_data, interprocess_dict,))
        processes.append(proc)
        proc.start()

    for proc in processes:
        proc.join()
    
    for user_id, user_data in interprocess_dict.items():
        print('{}: {}'.format(user_id, user_data))


if __name__ == '__main__':
    main()

EDIT:

Лучше ограничить количество количество процессов на количество доступных аппаратных модулей ЦП, поскольку сортировка списка является 100% операцией, связанной с ЦП.

import multiprocessing as mp


def sort_list(user_id, user_data, interprocess_dict):
    user_data.sort()
    interprocess_dict[user_id] = user_data


def prepare_data():
    users_data = {}
    for i in range(1000):
        users_data[i] = list(range(10000, 0, -1))
    return users_data


def main():
    # mp.set_start_method('spawn') # Only valid on OSX
    interprocess_dict = mp.Manager().dict()
    pool = mp.Pool(mp.cpu_count())
    users_data = prepare_data()
    for user_id, user_data in users_data.items():
        pool.apply_async(sort_list, args = (user_id, user_data, interprocess_dict,))
    pool.close()
    pool.join()
    for user_id, user_data in interprocess_dict.items():
        print('{}: {}'.format(user_id, user_data))


if __name__ == '__main__':
    main()