Используйте массив numpy в общей памяти для многопроцессорности

112

Я хотел бы использовать массив numpy в общей памяти для использования с модулем многопроцессорности. Сложность состоит в том, чтобы использовать его как массив numpy, а не только как массив ctypes.

from multiprocessing import Process, Array
import scipy

def f(a):
    a[0] = -a[0]

if __name__ == '__main__':
    # Create the array
    N = int(10)
    unshared_arr = scipy.rand(N)
    arr = Array('d', unshared_arr)
    print "Originally, the first two elements of arr = %s"%(arr[:2])

    # Create, start, and finish the child processes
    p = Process(target=f, args=(arr,))
    p.start()
    p.join()

    # Printing out the changed values
    print "Now, the first two elements of arr = %s"%arr[:2]

Это дает такой результат, как:

Originally, the first two elements of arr = [0.3518653236697369, 0.517794725524976]
Now, the first two elements of arr = [-0.3518653236697369, 0.517794725524976]

Доступ к массиву можно получить с помощью ctypes, например, arr[i]имеет смысл. Однако это не массив numpy, и я не могу выполнять такие операции, как -1*arr, или arr.sum(). Я полагаю, что решением было бы преобразовать массив ctypes в массив numpy. Однако (помимо того, что я не могу выполнить эту работу), я не верю, что ею больше будут делиться.

Кажется, было бы стандартное решение того, что должно быть общей проблемой.

Ян Лэнгмор
источник
1
Это не то же самое, что этот? stackoverflow.com/questions/5033799/…
pygabriel
1
Это не совсем тот же вопрос. Связанный вопрос subprocessскорее спрашивает , чем multiprocessing.
Эндрю

Ответы:

83

Чтобы добавить к ответам @ unutbu (больше не доступно) и @Henry Gomersall. Вы можете использовать shared_arr.get_lock()для синхронизации доступа при необходимости:

shared_arr = mp.Array(ctypes.c_double, N)
# ...
def f(i): # could be anything numpy accepts as an index such another numpy array
    with shared_arr.get_lock(): # synchronize access
        arr = np.frombuffer(shared_arr.get_obj()) # no data copying
        arr[i] = -arr[i]

пример

import ctypes
import logging
import multiprocessing as mp

from contextlib import closing

import numpy as np

info = mp.get_logger().info

def main():
    logger = mp.log_to_stderr()
    logger.setLevel(logging.INFO)

    # create shared array
    N, M = 100, 11
    shared_arr = mp.Array(ctypes.c_double, N)
    arr = tonumpyarray(shared_arr)

    # fill with random values
    arr[:] = np.random.uniform(size=N)
    arr_orig = arr.copy()

    # write to arr from different processes
    with closing(mp.Pool(initializer=init, initargs=(shared_arr,))) as p:
        # many processes access the same slice
        stop_f = N // 10
        p.map_async(f, [slice(stop_f)]*M)

        # many processes access different slices of the same array
        assert M % 2 # odd
        step = N // 10
        p.map_async(g, [slice(i, i + step) for i in range(stop_f, N, step)])
    p.join()
    assert np.allclose(((-1)**M)*tonumpyarray(shared_arr), arr_orig)

def init(shared_arr_):
    global shared_arr
    shared_arr = shared_arr_ # must be inherited, not passed as an argument

def tonumpyarray(mp_arr):
    return np.frombuffer(mp_arr.get_obj())

def f(i):
    """synchronized."""
    with shared_arr.get_lock(): # synchronize access
        g(i)

def g(i):
    """no synchronization."""
    info("start %s" % (i,))
    arr = tonumpyarray(shared_arr)
    arr[i] = -1 * arr[i]
    info("end   %s" % (i,))

if __name__ == '__main__':
    mp.freeze_support()
    main()

Если вам не нужен синхронизированный доступ или вы создаете свои собственные блокировки, в mp.Array()этом нет необходимости. Вы можете использовать mp.sharedctypes.RawArrayв этом случае.

JFS
источник
2
Прекрасный ответ! Если я хочу иметь более одного разделяемого массива, каждый отдельно блокируемый, но с количеством массивов, определенным во время выполнения, является ли это прямым расширением того, что вы здесь сделали?
Эндрю
3
@Andrew: общие массивы должны быть созданы до порождения дочерних процессов.
jfs
Хорошее замечание о порядке операций. Но это то, что я имел в виду: создать указанное пользователем количество общих массивов, а затем создать несколько дочерних процессов. Это просто?
Эндрю
1
@Chicony: вы не можете изменить размер массива. Думайте об этом как об общем блоке памяти, который должен был быть выделен до запуска дочерних процессов. Вам не нужно использовать всю память , например, вы могли бы перейти countк numpy.frombuffer(). Вы можете попробовать сделать это на более низком уровне, используя mmapили что-то вроде того, posix_ipcчтобы реализовать изменяемый размер (может включать копирование при изменении размера) аналог RawArray (или искать существующую библиотеку). Или, если ваша задача позволяет: копировать данные по частям (если вам не нужно все сразу). «Как изменить размер разделяемой памяти» - хороший отдельный вопрос.
jfs
1
@umopapisdn: Pool()определяет количество процессов (по умолчанию используется количество доступных ядер процессора). M- количество вызовов f()функции.
jfs
21

С Arrayобъектом get_obj()связан метод, который возвращает массив ctypes, представляющий интерфейс буфера. Я думаю, что следующее должно работать ...

from multiprocessing import Process, Array
import scipy
import numpy

def f(a):
    a[0] = -a[0]

if __name__ == '__main__':
    # Create the array
    N = int(10)
    unshared_arr = scipy.rand(N)
    a = Array('d', unshared_arr)
    print "Originally, the first two elements of arr = %s"%(a[:2])

    # Create, start, and finish the child process
    p = Process(target=f, args=(a,))
    p.start()
    p.join()

    # Print out the changed values
    print "Now, the first two elements of arr = %s"%a[:2]

    b = numpy.frombuffer(a.get_obj())

    b[0] = 10.0
    print a[0]

При запуске, это выводит на первый элемент в aнастоящее время 10.0, показывая aи bлишь два мнения в той же памяти.

Для того , чтобы убедиться , что он по - прежнему многопроцессорных безопасности, я считаю , вы должны будете использовать acquireи releaseметоды , которые существуют на Arrayобъекте, aи встроенный в замок , чтобы убедиться , что его все благополучно получить доступ (хотя я не эксперт на многопроцессорный модуль).

Генри Гомерсолл
источник
он не будет работать без синхронизации, как продемонстрировал @unutbu в своем (теперь удаленном) ответе.
jfs
1
Предположительно, если вы просто хотите получить доступ к постобработке массива, это можно сделать чисто, не беспокоясь о проблемах параллелизма и блокировках?
Генри Гомерсалл,
в этом случае вам не нужно mp.Array.
jfs
1
Для кода обработки могут потребоваться заблокированные массивы, но интерпретация данных после обработки может не потребоваться. Я думаю, это происходит от понимания, в чем именно проблема. Ясно, что одновременный доступ к общим данным потребует некоторой защиты, что, как я думал, было очевидно!
Генри Гомерсалл
16

Хотя уже даны хорошие ответы, существует гораздо более простое решение этой проблемы при соблюдении двух условий:

  1. Вы работаете в POSIX-совместимой операционной системе (например, Linux, Mac OSX); и
  2. Вашим дочерним процессам требуется доступ только для чтения к общему массиву.

В этом случае вам не нужно возиться с явным созданием общих переменных, поскольку дочерние процессы будут созданы с помощью вилки. Разветвленный дочерний элемент автоматически разделяет пространство памяти родителя. В контексте многопроцессорной обработки Python это означает, что он разделяет все переменные уровня модуля ; обратите внимание, что это не относится к аргументам, которые вы явно передаете своим дочерним процессам, или к функциям, которые вы вызываете на каком-то multiprocessing.Poolили около того.

Простой пример:

import multiprocessing
import numpy as np

# will hold the (implicitly mem-shared) data
data_array = None

# child worker function
def job_handler(num):
    # built-in id() returns unique memory ID of a variable
    return id(data_array), np.sum(data_array)

def launch_jobs(data, num_jobs=5, num_worker=4):
    global data_array
    data_array = data

    pool = multiprocessing.Pool(num_worker)
    return pool.map(job_handler, range(num_jobs))

# create some random data and execute the child jobs
mem_ids, sumvals = zip(*launch_jobs(np.random.rand(10)))

# this will print 'True' on POSIX OS, since the data was shared
print(np.all(np.asarray(mem_ids) == id(data_array)))
EelkeSpaak
источник
3
+1 Действительно ценная информация. Можете ли вы объяснить, почему используются только переменные уровня модуля? Почему локальные вары не являются частью родительского пространства памяти? Например, почему это не может работать, если у меня есть функция F с локальной переменной V и функция G внутри F, которая ссылается на V?
Coffee_Table
5
Предупреждение: этот ответ немного обманчив. Дочерний процесс получает копию состояния родительского процесса, включая глобальные переменные, во время вилки. Состояния никоим образом не синхронизированы и с этого момента будут расходиться. Этот метод может быть полезен в некоторых сценариях (например: разветвление специальных дочерних процессов, каждый из которых обрабатывает моментальный снимок родительского процесса, а затем завершается), но бесполезен в других (например: длительные дочерние процессы, которые должны совместно использовать и синхронизировать данные с родительским процессом).
Дэвид Штайн
4
@EelkeSpaak: Ваше утверждение - «разветвленный дочерний элемент автоматически разделяет пространство памяти родителя» - неверно. Если у меня есть дочерний процесс, который хочет отслеживать состояние родительского процесса строго в режиме «только для чтения», разветвление не приведет меня туда: дочерний процесс видит только снимок родительского состояния в момент разветвления. Фактически, это именно то, что я пытался сделать (следуя вашему ответу), когда обнаружил это ограничение. Отсюда постскриптум к вашему ответу. В двух словах: родительское состояние не «разделяется», а просто копируется в дочернее состояние. Это не «делиться» в обычном смысле.
Дэвид Штейн
2
Я ошибаюсь, думая, что это ситуация копирования при записи, по крайней мере, в системах posix? То есть после вилки я думаю, что память используется совместно до тех пор, пока не будут записаны новые данные, после чего создается копия. Так что да, это правда, что данные точно не «разделяются», но они могут обеспечить потенциально огромный прирост производительности. Если ваш процесс доступен только для чтения, накладных расходов на копирование не будет! Правильно ли я понял суть?
senderle
2
@senderle Да, я именно это имел в виду! Отсюда мой пункт (2) в ответе о доступе только для чтения.
EelkeSpaak 01
11

Я написал небольшой модуль python, который использует разделяемую память POSIX для совместного использования массивов numpy между интерпретаторами python. Может быть, тебе это пригодится.

https://pypi.python.org/pypi/SharedArray

Вот как это работает:

import numpy as np
import SharedArray as sa

# Create an array in shared memory
a = sa.create("test1", 10)

# Attach it as a different array. This can be done from another
# python interpreter as long as it runs on the same computer.
b = sa.attach("test1")

# See how they are actually sharing the same memory block
a[0] = 42
print(b[0])

# Destroying a does not affect b.
del a
print(b[0])

# See how "test1" is still present in shared memory even though we
# destroyed the array a.
sa.list()

# Now destroy the array "test1" from memory.
sa.delete("test1")

# The array b is not affected, but once you destroy it then the
# data are lost.
print(b[0])
мат
источник
8

Вы можете использовать sharedmemмодуль: https://bitbucket.org/cleemesser/numpy-sharedmem

Вот ваш исходный код, на этот раз с использованием общей памяти, которая ведет себя как массив NumPy (обратите внимание на дополнительный последний оператор, вызывающий функцию NumPy sum()):

from multiprocessing import Process
import sharedmem
import scipy

def f(a):
    a[0] = -a[0]

if __name__ == '__main__':
    # Create the array
    N = int(10)
    unshared_arr = scipy.rand(N)
    arr = sharedmem.empty(N)
    arr[:] = unshared_arr.copy()
    print "Originally, the first two elements of arr = %s"%(arr[:2])

    # Create, start, and finish the child process
    p = Process(target=f, args=(arr,))
    p.start()
    p.join()

    # Print out the changed values
    print "Now, the first two elements of arr = %s"%arr[:2]

    # Perform some NumPy operation
    print arr.sum()
Велимир Млакер
источник
1
Примечание: это больше не разрабатывается и, похоже, не работает в linux github.com/sturlamolden/sharedmem-numpy/issues/4
AD
numpy-sharedmem может не находиться в разработке, но он все еще работает в Linux, проверьте github.com/vmlaker/benchmark-sharedmem .
Велимир Млакер