multiprocessing.shared_memory — Общая память для прямого доступа между процессами

Исходный код: Lib/multiprocessing/shared_memory.py

Этот модуль предоставляет класс SharedMemory для выделения и управления общей памятью, к которой могут обращаться один или несколько процессов на многоядерной машине или машине с симметричной многопроцессорной архитектурой (SMP). Для упрощения управления жизненным циклом общей памяти, особенно между отдельными процессами, в модуле multiprocessing.managers также предоставлен подкласс BaseManager — SharedMemoryManager.

В этом модуле под общей памятью подразумеваются блоки общей памяти «в стиле POSIX» (хотя они не обязательно реализованы именно таким образом); это понятие не относится к «распределённой общей памяти». Такой тип общей памяти позволяет отдельным процессам читать и записывать данные в общую область энергозависимой памяти. Обычно процессы имеют доступ только к собственной области памяти, однако общая память позволяет обмениваться данными между процессами и избавляет от необходимости отправлять друг другу сообщения с этими данными. Прямой обмен данными через память может обеспечить значительный выигрыш в производительности по сравнению с обменом данными через диск, сокет или другие средства связи, требующие сериализации, десериализации и копирования данных.

class multiprocessing.shared_memory.SharedMemory(name=None, create=False, size=0)

Создаёт экземпляр класса SharedMemory для создания нового блока общей памяти или подключения к существующему блоку. Каждому блоку общей памяти присваивается уникальное имя. Благодаря этому один процесс может создать блок общей памяти с определённым именем, а другой процесс — подключиться к тому же блоку, используя это имя.

Блоки общей памяти, предназначенные для обмена данными между процессами, могут существовать дольше процесса, который их создал. Если процессу больше не нужен доступ к блоку общей памяти, который всё ещё может понадобиться другим процессам, следует вызвать метод close(). Когда блок общей памяти больше не нужен ни одному процессу, следует вызвать метод unlink(), чтобы обеспечить корректную очистку.

Параметры:

• name (str | None) – Уникальное имя запрашиваемой общей памяти, заданное в виде строки. При создании нового блока общей памяти, если для имени указано None (значение по умолчанию), будет сгенерировано новое имя.
• create (bool) – Определяет, нужно ли создать новый блок общей памяти (True) или подключиться к существующему (False).
• size (int) – Запрашиваемое количество байтов при создании нового блока общей памяти. На некоторых платформах память выделяется блоками, размер которых определяется размером страницы памяти этой платформы, поэтому фактический размер блока общей памяти может быть больше запрошенного или равен ему. При подключении к существующему блоку общей памяти параметр size игнорируется.

close()

Закрывает доступ к общей памяти из этого экземпляра. Для корректной очистки ресурсов все экземпляры должны вызывать close(), когда экземпляр больше не нужен. Обратите внимание: вызов close() не приводит к уничтожению самого блока общей памяти.

unlink()

Запрашивает уничтожение соответствующего блока общей памяти. Для корректной очистки ресурсов unlink() следует вызвать один раз (и только один раз) во всех процессах, которым нужен этот блок общей памяти. После запроса на уничтожение блок общей памяти может быть уничтожен как сразу, так и позже; поведение может различаться на разных платформах. Попытки доступа к данным в блоке общей памяти после вызова unlink() могут привести к ошибкам доступа к памяти. Примечание: последний процесс, прекращающий использование блока общей памяти, может вызвать unlink() и close() в любом порядке.

buf

Представление памяти содержимого блока общей памяти.

name

Доступ только для чтения к уникальному имени блока общей памяти.

size

Доступ только для чтения к размеру блока общей памяти в байтах.

В следующем примере показано низкоуровневое использование экземпляров SharedMemory:

>>> from multiprocessing import shared_memory
>>> shm_a = shared_memory.SharedMemory(create=True, size=10)
>>> type(shm_a.buf)
<class 'memoryview'>
>>> buffer = shm_a.buf
>>> len(buffer)
10
>>> buffer[:4] = bytearray([22, 33, 44, 55])  # Modify multiple at once
>>> buffer[4] = 100                           # Modify single byte at a time
>>> # Attach to an existing shared memory block
>>> shm_b = shared_memory.SharedMemory(shm_a.name)
>>> import array
>>> array.array('b', shm_b.buf[:5])  # Copy the data into a new array.array
array('b', [22, 33, 44, 55, 100])
>>> shm_b.buf[:5] = b'howdy'  # Modify via shm_b using bytes
>>> bytes(shm_a.buf[:5])      # Access via shm_a
b'howdy'
>>> shm_b.close()   # Close each SharedMemory instance
>>> shm_a.close()
>>> shm_a.unlink()  # Call unlink only once to release the shared memory

В следующем примере показано практическое использование класса SharedMemory с массивами NumPy для доступа к одной и той же numpy.ndarray из двух разных оболочек Python:

>>> # In the first Python interactive shell
>>> import numpy as np
>>> a = np.array([1, 1, 2, 3, 5, 8])  # Start with an existing NumPy array
>>> from multiprocessing import shared_memory
>>> shm = shared_memory.SharedMemory(create=True, size=a.nbytes)
>>> # Now create a NumPy array backed by shared memory
>>> b = np.ndarray(a.shape, dtype=a.dtype, buffer=shm.buf)
>>> b[:] = a[:]  # Copy the original data into shared memory
>>> b
array([1, 1, 2, 3, 5, 8])
>>> type(b)
<class 'numpy.ndarray'>
>>> type(a)
<class 'numpy.ndarray'>
>>> shm.name  # We did not specify a name so one was chosen for us
'psm_21467_46075'

>>> # In either the same shell or a new Python shell on the same machine
>>> import numpy as np
>>> from multiprocessing import shared_memory
>>> # Attach to the existing shared memory block
>>> existing_shm = shared_memory.SharedMemory(name='psm_21467_46075')
>>> # Note that a.shape is (6,) and a.dtype is np.int64 in this example
>>> c = np.ndarray((6,), dtype=np.int64, buffer=existing_shm.buf)
>>> c
array([1, 1, 2, 3, 5, 8])
>>> c[-1] = 888
>>> c
array([  1,   1,   2,   3,   5, 888])

>>> # Back in the first Python interactive shell, b reflects this change
>>> b
array([  1,   1,   2,   3,   5, 888])

>>> # Clean up from within the second Python shell
>>> del c  # Unnecessary; merely emphasizing the array is no longer used
>>> existing_shm.close()

>>> # Clean up from within the first Python shell
>>> del b  # Unnecessary; merely emphasizing the array is no longer used
>>> shm.close()
>>> shm.unlink()  # Free and release the shared memory block at the very end

class multiprocessing.managers.SharedMemoryManager([address[, authkey]])

Подкласс multiprocessing.managers.BaseManager, предназначенный для управления блоками общей памяти между процессами.

Вызов start() для экземпляра SharedMemoryManager приводит к запуску нового процесса. Единственная задача этого нового процесса — управлять жизненным циклом всех блоков общей памяти, созданных через него. Чтобы освободить все блоки общей памяти, которыми управляет этот процесс, вызовите для экземпляра shutdown(). Это приводит к вызову unlink() для всех объектов SharedMemory, которыми управляет этот процесс, после чего сам процесс останавливается. Создавая экземпляры SharedMemory с помощью SharedMemoryManager, мы избавляемся от необходимости вручную отслеживать и запускать освобождение ресурсов общей памяти.

Этот класс предоставляет методы для создания и возврата экземпляров SharedMemory, а также для создания объекта, похожего на список (ShareableList), данные которого хранятся в общей памяти.

Описание унаследованных необязательных аргументов address и authkey, а также способов их использования для подключения к существующей службе SharedMemoryManager из других процессов см. в документации по BaseManager.

SharedMemory(size)

Создаёт и возвращает новый объект SharedMemory заданного размера size в байтах.

ShareableList(sequence)

Создаёт и возвращает новый объект ShareableList, заполненный значениями из входной sequence.

В следующем примере показаны основные механизмы работы SharedMemoryManager:

>>> from multiprocessing.managers import SharedMemoryManager
>>> smm = SharedMemoryManager()
>>> smm.start()  # Start the process that manages the shared memory blocks
>>> sl = smm.ShareableList(range(4))
>>> sl
ShareableList([0, 1, 2, 3], name='psm_6572_7512')
>>> raw_shm = smm.SharedMemory(size=128)
>>> another_sl = smm.ShareableList('alpha')
>>> another_sl
ShareableList(['a', 'l', 'p', 'h', 'a'], name='psm_6572_12221')
>>> smm.shutdown()  # Calls unlink() on sl, raw_shm, and another_sl

В следующем примере показан потенциально более удобный способ использования объектов SharedMemoryManager с помощью инструкции with, которая гарантирует освобождение всех блоков общей памяти после того, как они перестанут быть нужны:

>>> with SharedMemoryManager() as smm:
...     sl = smm.ShareableList(range(2000))
...     # Divide the work among two processes, storing partial results in sl
...     p1 = Process(target=do_work, args=(sl, 0, 1000))
...     p2 = Process(target=do_work, args=(sl, 1000, 2000))
...     p1.start()
...     p2.start()  # A multiprocessing.Pool might be more efficient
...     p1.join()
...     p2.join()   # Wait for all work to complete in both processes
...     total_result = sum(sl)  # Consolidate the partial results now in sl

При использовании SharedMemoryManager в инструкции with все созданные с помощью этого менеджера блоки общей памяти освобождаются по завершении выполнения блока кода инструкции with.

class multiprocessing.shared_memory.ShareableList(sequence=None, *, name=None)

Предоставляет изменяемый объект, похожий на список, в котором все значения хранятся в блоке общей памяти. Это ограничивает допустимые для хранения значения следующими встроенными типами данных:

• int (знаковое 64-разрядное целое число)
• float
• bool
• str (не более 10 млн байт каждое при кодировании в UTF-8)
• bytes (не более 10 млн байт каждое)
• None

Также этот тип заметно отличается от встроенного типа list: такие списки не могут менять свою общую длину (то есть не поддерживают append(), insert() и т. д.) и не позволяют динамически создавать новые экземпляры ShareableList с помощью срезов.

sequence используется для заполнения нового ShareableList значениями. Укажите None, чтобы вместо этого подключиться к уже существующему ShareableList по его уникальному имени в общей памяти.

name — уникальное имя запрашиваемой общей памяти, как описано в определении SharedMemory. При подключении к существующему ShareableList укажите уникальное имя его блока общей памяти, оставив для sequence значение None.

Примечание

Известна проблема со значениями типа bytes и str. Если они заканчиваются \x00 нулевыми байтами или символами, при извлечении по индексу из ShareableList они могут быть незаметно удалены. Такое поведение .rstrip(b'\x00') считается ошибкой и в будущем может быть исправлено. См. gh-106939.

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

>>> from multiprocessing import shared_memory
>>> nul_bug_demo = shared_memory.ShareableList(['?\x00', b'\x03\x02\x01\x00\x00\x00'])
>>> nul_bug_demo[0]
'?'
>>> nul_bug_demo[1]
b'\x03\x02\x01'
>>> nul_bug_demo.shm.unlink()
>>> padded = shared_memory.ShareableList(['?\x00\x07', b'\x03\x02\x01\x00\x00\x00\x07'])
>>> padded[0][:-1]
'?\x00'
>>> padded[1][:-1]
b'\x03\x02\x01\x00\x00\x00'
>>> padded.shm.unlink()

count(value)

Возвращает количество вхождений значения value.

index(value)

Возвращает индекс первого вхождения значения value. Если value отсутствует, вызывает исключение ValueError.

format

Атрибут только для чтения, содержащий формат упаковки struct, используемый для всех хранящихся в данный момент значений.

shm

Экземпляр SharedMemory, в котором хранятся значения.

В следующем примере показано базовое использование экземпляра ShareableList:

>>> from multiprocessing import shared_memory
>>> a = shared_memory.ShareableList(['howdy', b'HoWdY', -273.154, 100, None, True, 42])
>>> [ type(entry) for entry in a ]
[<class 'str'>, <class 'bytes'>, <class 'float'>, <class 'int'>, <class 'NoneType'>, <class 'bool'>, <class 'int'>]
>>> a[2]
-273.154
>>> a[2] = -78.5
>>> a[2]
-78.5
>>> a[2] = 'dry ice'  # Changing data types is supported as well
>>> a[2]
'dry ice'
>>> a[2] = 'larger than previously allocated storage space'
Traceback (most recent call last):
  ...
ValueError: exceeds available storage for existing str
>>> a[2]
'dry ice'
>>> len(a)
7
>>> a.index(42)
6
>>> a.count(b'howdy')
0
>>> a.count(b'HoWdY')
1
>>> a.shm.close()
>>> a.shm.unlink()
>>> del a  # Use of a ShareableList after call to unlink() is unsupported

В следующем примере показано, как один, два или несколько процессов могут обращаться к одному и тому же объекту ShareableList, указав имя лежащего в его основе блока общей памяти:

>>> b = shared_memory.ShareableList(range(5))         # In a first process
>>> c = shared_memory.ShareableList(name=b.shm.name)  # In a second process
>>> c
ShareableList([0, 1, 2, 3, 4], name='...')
>>> c[-1] = -999
>>> b[-1]
-999
>>> b.shm.close()
>>> c.shm.close()
>>> c.shm.unlink()

В следующих примерах показано, что объекты ShareableList (и лежащие в их основе объекты SharedMemory) при необходимости можно сериализовать с помощью pickle и десериализовать. Обратите внимание, что это всё равно будет тот же общий объект. Так происходит потому, что десериализованный объект имеет то же уникальное имя и просто подключается к существующему объекту с таким именем (если объект всё ещё существует):

>>> import pickle
>>> from multiprocessing import shared_memory
>>> sl = shared_memory.ShareableList(range(10))
>>> list(sl)
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

>>> deserialized_sl = pickle.loads(pickle.dumps(sl))
>>> list(deserialized_sl)
[0, 1, 2, 3, 4, 5, 6, 7, 8, 9]

>>> sl[0] = -1
>>> deserialized_sl[1] = -2
>>> list(sl)
[-1, -2, 2, 3, 4, 5, 6, 7, 8, 9]
>>> list(deserialized_sl)
[-1, -2, 2, 3, 4, 5, 6, 7, 8, 9]

>>> sl.shm.close()
>>> sl.shm.unlink()