Изменение объекта в многопроцессорности python

Ответ 1

Проблема заключается в том, что когда объекты передаются рабочим процессам, они упаковываются с рассолом, отправляются в другой процесс, где они распаковываются и обрабатываются. Ваши объекты не столько передаются другому процессу, сколько клонированы. Вы не возвращаете объекты, поэтому клонированный объект с радостью модифицируется, а затем отбрасывается.

Похоже, это невозможно сделать (Python: возможность совместного использования данных в памяти между двумя отдельными процессами).

Что вы можете сделать, это вернуть измененные объекты.

import numpy as np
import multiprocessing as mp



class Tester:

    num = 0.0
    name = 'none'
    def __init__(self,tnum=num, tname=name):
        self.num  = tnum
        self.name = tname

    def __str__(self):
        return '%f %s' % (self.num, self.name)

def mod(test, nn, out_queue):
    print test.num
    test.num = np.random.randn()
    print test.num
    test.name = nn
    out_queue.put(test)




if __name__ == '__main__':       
    num = 10
    out_queue = mp.Queue()
    tests = np.empty(num, dtype=object)
    for it in range(num):
        tests[it] = Tester(tnum=it*1.0)


    print '\n'
    workers = [ mp.Process(target=mod, args=(test, 'some', out_queue) ) for test in tests ]

    for work in workers: work.start()

    for work in workers: work.join()

    res_lst = []
    for j in range(len(workers)):
        res_lst.append(out_queue.get())

    for test in res_lst: print test

Это приводит к интересному замечанию о том, что, поскольку порожденные процессы идентичны, все они начинаются с одного и того же семени для случайного числа, поэтому все они генерируют одно и то же "случайное" число.

Ответ 2

Я не вижу, чтобы вы передавали ссылки shm в дочерние процессы, поэтому я не вижу, как выполненная ими работа может быть записана обратно в разделяемую память. Возможно, я что-то пропустил.

В качестве альтернативы вы рассмотрели numpy.memmap? (BTW: tcaswell, упомянутый здесь модуль выглядит следующим образом: numpy-sharedmem).

Также вы можете захотеть прочитать Sturla Molden Использование Python, многопроцессорности и NumPy/SciPy для параллельных вычислительных вычислений (PDF), как рекомендовано в unutbu ответ на [StackOverflow: Как передать большие массивы numpy между подпроцессами python без сохранения на диск?] и (Как передать большие массивы numpy между подпроцессами python без сохранения на диск?). и Joe Kington fooobar.com/questions/287723/....

Они могут быть более вдохновляющими, чем прямо релевантные.

Ответ 3

Ваш код не пытается изменить общую память. Он просто клонирует отдельные объекты.

dtype=object означает, что sharedmem не будет работать по причинам, указанным в в ссылке, предоставленной @tcaswell:

Обмен объектных графов, содержащих ссылки/указатели на другие объекты, в основном невыполним

Для простых (значений) типов вы можете использовать общую память, см. Использовать массив numpy в общей памяти для многопроцессорности.

Подход manager также должен работать (он просто копирует объекты):

import random
from multiprocessing import Pool, Manager

class Tester(object):
    def __init__(self, num=0.0, name='none'):
        self.num  = num
        self.name = name

    def __repr__(self):
        return '%s(%r, %r)' % (self.__class__.__name__, self.num, self.name)

def init(L):
    global tests
    tests = L

def modify(i_t_nn):
    i, t, nn = i_t_nn
    t.num += random.normalvariate(mu=0, sigma=1) # modify private copy
    t.name = nn
    tests[i] = t # copy back
    return i

def main():
    num_processes = num = 10 #note: num_processes and num may differ
    manager = Manager()
    tests = manager.list([Tester(num=i) for i in range(num)])
    print(tests[:2])

    args = ((i, t, 'some') for i, t in enumerate(tests))
    pool = Pool(processes=num_processes, initializer=init, initargs=(tests,))
    for i in pool.imap_unordered(modify, args):
        print("done %d" % i)
    pool.close()
    pool.join()
    print(tests[:2])

if __name__ == '__main__':
    main()

Ответ 4

Поскольку вы не можете совместно использовать объекты Python между процессами, любая реализация, использующая multiprocessing будет неэффективной, если у вас есть важные объекты, поскольку вам придется копировать объекты для обмена данными.

Если вы хотите попробовать другой подход, вы можете попробовать Ray (документы)! Это фреймворк, который облегчает написание параллельного и распределенного Python. В двух словах, он дает вам возможность запускать функции Python параллельно, аналогично multiprocessing, но он также более гибок в том, что процессы Ray могут совместно использовать память. Вот ваш сценарий, написанный на Ray, с использованием понятия "актеры" (разделяемые объекты):

# You can install Ray with pip.
import ray

import numpy as np


# Add this line to signify that you want to share Tester objects
# (called "actors" in Ray) between processes.
@ray.remote
class Tester(object):

    num = 0.0
    name = 'none'
    def __init__(self,tnum=num, tname=name):
        self.num  = tnum
        self.name = tname

    def __str__(self):
        return '%f %s' % (self.num, self.name)

    # Convert mod to be a method of the Tester object.
    def mod(self, nn):
        self.num = np.random.randn()
        self.name = nn


if __name__ == '__main__':

    # Start Ray. This allows you to create shared Testers (called "actors").
    ray.init()

    num = 10

    tests = np.empty(num, dtype=object)
    for it in range(num):
        # Create a shared Tester object (an "actor").
        tests[it] = Tester.remote(tnum=it*1.0)

    # Do some parallel work.
    for test in tests:
        test.mod.remote('some')

    # Compute the __str__ representations of each Tester in parallel.
    test_str_futures = [test.__str__.remote() for test in tests]
    # Get and print the __str__ return values. 'ray.get' will block
    # until the return values are ready.
    test_strs = ray.get(test_str_futures)
    for test_str in test_strs:
        print(test_str)