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

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

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 значениями. Чтобы вместо этого подключиться к уже существующему ShareableList по его уникальному имени, заданному для разделяемой памяти, укажите None.

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

Примечание

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

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

>>> 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()