Как в Python3 использовать вложенный список в multiprocessing?

Question

[ARTIsshoque]

К примеру, у меня есть словарь:

D = {'x': [1, 2, 3], 'y': [4, 5, 6], 'z': [7, 8, 9]}

и я хочу обработать его, используя несколько процессов. Для упрощения примера, пусть нужно просто умножить каждый элемент списков на 10 (хотя в реальности будет использоваться более сложная функция, для которой мультипроцессинг целесообразен). Набросок кода такой:

import multiprocessing

PROCESSES = 4

class Worker(multiprocessing.Process):

    def __init__(self, work_queue):
        super().__init__()
        self.work_queue = work_queue

    def run(self):
        while True:
            try:
                key, index = self.work_queue.get()
                self.process(key, index)
            finally:
                self.work_queue.task_done()

    def process(self, key, index):
        D[key][index] = D[key][index] * 10
        # Здесь написана полная ерунда, но я просто хочу показать, что мне нужно получить


def main():
    work_queue = multiprocessing.JoinableQueue()
    for i in range(PROCESSES):
        worker = Worker(work_queue)
        worker.daemon = True
        worker.start()
    for key, value in D.items():
        for i in range(len(value)):
            work_queue.put((key, i))
    work_queue.join()

main()

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

Comments

[longclaps]

Может, стоит освежить свои знания о mutable?

Answer 1

[Roman K]

Не надо так низкоуровнево писать (обычно).

from concurrent.futures import ProcessPoolExecutor


D = {'x': [1, 2, 3], 'y': [4, 5, 6], 'z': [7, 8, 9]}


def process(arg):
    key, values = arg
    return key, [v * 10 for v in values]


if __name__ == "__main__":
    with ProcessPoolExecutor() as executor:
        result = executor.map(process, D.items())

    print(dict(result))

Чтобы убедиться, что оно действительно работает и не блокирует друг друга:

from concurrent.futures import ProcessPoolExecutor
from time import sleep


D = {'x': [1, 2, 3], 'y': [4, 5, 6], 'z': [7, 8, 9]}


def process(arg):
    key, values = arg
    print("executing", key, values)
    sleep(1)
    return key, [v * 10 for v in values]


if __name__ == "__main__":
    with ProcessPoolExecutor(2) as executor:
        result = executor.map(process, D.items())

    print(dict(result))

Comments

[ARTIsshoque]

Это хорошо, но в вашем примере между процессами делится словарь, а список целиком обрабатывается в одном процессе. Это немного не то, что нужно.

[Roman K]

ARTIsshoque, он не делится. В процессы передаются пары ключ-значение. Условие передачи одно - структура должна уметь pickle.dump

[Roman K]

ARTIsshoque, можешь засунуть словарь в __main__ и убедиться сам

[ARTIsshoque]

Pavel Denisov, в приведённом примере я сильно упростил задачу. На самом деле процессу нужен и ключ словаря, и индекс списка. То есть, значение получается не просто умножением 1, 2, 3 на 10, а 'x' в этом тоже участвует. Другими словами, мне нужен аналог D[key][index] = func(key, D[key][index]), разделённый на процессы. И списки могут состоять не из трёх элементов, поэтому я не могу представить, как эту структуру привести к одному списку.

[ARTIsshoque]

Pavel Denisov, думаю, стоит попробовать. Жаль, конечно, что нельзя использовать сложную структуру напрямую, но ваш вариант + Manager().list() - пока лучшее из всех возможных решений.

[Roman K]

Pavel Denisov, ну так это же не проблема. Нужно всего лишь переписать мой код под его нужды. Если ТС и этого делать не хочет — тут я бессилен.

[ARTIsshoque]

Roman Kitaev, получилось пока примерно так:

from concurrent.futures import ProcessPoolExecutor
from time import sleep


def process(arg):
    key, index, value = arg
    print("executing", key, value)
    sleep(1)
    return key, index, value * 10


if __name__ == "__main__":
    D = {'x': [1, 2, 3], 'y': [4, 5, 6], 'z': [7, 8, 9]}
    L = []
    for key, value in D.items():
        for index in range(len(value)):
            L.append((key, index, value[index]))
    with ProcessPoolExecutor(2) as executor:
        result = executor.map(process, L)
    for x, y, z in result:
        D[x][y] = z

    print(D)