multiprocessing.shared_memory – Разделяемая память для прямого доступа между процессами¶multiprocessing.shared_memory – Shared memory for direct access across processes

Исходный код: 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])  # Изменить несколько за раз
>>> buffer[4] = 100                           # Изменить один байт за раз
>>> # Подключиться к существующему блоку разделяемой памяти
>>> shm_b = shared_memory.SharedMemory(shm_a.name)
>>> import array
>>> array.array('b', shm_b.buf[:5])  # Скопировать данные в новый array.array
array('b', [22, 33, 44, 55, 100])
>>> shm_b.buf[:5] = b'howdy'  # Изменить через shm_b, используя bytes
>>> bytes(shm_a.buf[:5])      # Доступ через shm_a
b'howdy'
>>> shm_b.close()   # Закрыть каждый экземпляр SharedMemory
>>> shm_a.close()
>>> shm_a.unlink()  # Вызвать unlink только один раз, чтобы освободить разделяемую память

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

>>> # В первой интерактивной оболочке Python
>>> import numpy as np
>>> a = np.array([1, 1, 2, 3, 5, 8])  # Начать с существующего массива NumPy
>>> from multiprocessing import shared_memory
>>> shm = shared_memory.SharedMemory(create=True, size=a.nbytes)
>>> # Теперь создать массив NumPy, поддерживаемый разделяемой памятью
>>> b = np.ndarray(a.shape, dtype=a.dtype, buffer=shm.buf)
>>> b[:] = a[:]  # Скопировать исходные данные в разделяемую память
>>> b
array([1, 1, 2, 3, 5, 8])
>>> type(b)
<class 'numpy.ndarray'>
>>> type(a)
<class 'numpy.ndarray'>
>>> shm.name  # Мы не указали имя, поэтому оно было выбрано автоматически
'psm_21467_46075'

>>> # В той же оболочке или в новой оболочке Python на той же машине
>>> import numpy as np
>>> from multiprocessing import shared_memory
>>> # Подключиться к существующему блоку разделяемой памяти
>>> existing_shm = shared_memory.SharedMemory(name='psm_21467_46075')
>>> # Обратите внимание, что a.shape равно (6,), а a.dtype равно np.int64 в этом примере
>>> 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])

>>> # Вернувшись в первую интерактивную оболочку Python, b отражает это изменение
>>> b
array([  1,   1,   2,   3,   5, 888])

>>> # Очистка из второй оболочки Python
>>> del c  # Необязательно; просто подчёркивается, что массив больше не используется
>>> existing_shm.close()

>>> # Очистка из первой оболочки Python
>>> del b  # Необязательно; просто подчёркивается, что массив больше не используется
>>> shm.close()
>>> shm.unlink()  # Освободить и удалить блок разделяемой памяти в самом конце

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

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

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

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

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

SharedMemory(size)¶

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

ShareableList(sequence)¶

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

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

>>> from multiprocessing.managers import SharedMemoryManager
>>> smm = SharedMemoryManager()
>>> smm.start()  # Запуск процесса, управляющего блоками разделяемой памяти
>>> 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()  # Вызывает unlink() для sl, raw_shm и another_sl

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

>>> with SharedMemoryManager() as smm:
...     sl = smm.ShareableList(range(2000))
...     # Разделение работы между двумя процессами, сохранение частичных результатов в sl
...     p1 = Process(target=do_work, args=(sl, 0, 1000))
...     p2 = Process(target=do_work, args=(sl, 1000, 2000))
...     p1.start()
...     p2.start()  # multiprocessing.Pool может быть эффективнее
...     p1.join()
...     p2.join()   # Ожидание завершения всей работы в обоих процессах
...     total_result = sum(sl)  # Объединение частичных результатов из 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.

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

>>> 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. Возбуждает ValueError, если value отсутствует.

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'  # Изменение типов данных также поддерживается
>>> 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  # Использование ShareableList после вызова unlink() не поддерживается

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

>>> b = shared_memory.ShareableList(range(5))         # В первом процессе
>>> c = shared_memory.ShareableList(name=b.shm.name)  # Во втором процессе
>>> 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) могут быть сериализованы и десериализованы при необходимости. Обратите внимание, что это всё равно будет тот же самый общий объект. Это происходит потому, что десериализованный объект имеет то же уникальное имя и просто подключается к существующему объекту с тем же именем (если объект ещё жив):

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