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, *, track=True)

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

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

Параметры:

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

Изменено в версии 3.13: Добавлен параметр track.

close()

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

unlink()

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

Этот метод не имеет эффекта в Windows, где единственный способ удалить блок общей памяти — закрыть все дескрипторы.

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 заданного размера в байтах.

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. Вызовите 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'  # 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) можно сериализовать и десериализовать при необходимости. Обратите внимание, что это по-прежнему будет тот же общий объект. Это происходит, потому что десериализованный объект имеет то же самое уникальное имя и просто подключается к существующему объекту с тем же именем (если объект все еще жив):

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