Skip to content

Instantly share code, notes, and snippets.

@wgsjack199213
Forked from jakirkham/shm.py
Created December 20, 2022 17:52
Show Gist options
  • Save wgsjack199213/329a089cbc931872ed8babee41b72e61 to your computer and use it in GitHub Desktop.
Save wgsjack199213/329a089cbc931872ed8babee41b72e61 to your computer and use it in GitHub Desktop.
Some Python ctypes-based POSIX shared memory functions
import ctypes
import mmap
import os
import stat
import sys
try:
unicode
except NameError:
unicode = str
rtld = ctypes.cdll.LoadLibrary(None)
_shm_open = rtld.shm_open
_shm_unlink = rtld.shm_unlink
def shm_open(name):
if isinstance(name, bytes):
name = ctypes.create_string_buffer(name)
elif isinstance(name, unicode):
name = ctypes.create_unicode_buffer(name)
else:
raise TypeError("`name` must be `bytes` or `unicode`")
result = _shm_open(
name,
ctypes.c_int(os.O_RDWR | os.O_CREAT | os.O_EXCL),
ctypes.c_ushort(stat.S_IRUSR | stat.S_IWUSR)
)
if result == -1:
raise RuntimeError(os.strerror(ctypes.get_errno()))
return result
def shm_unlink(name):
if isinstance(name, bytes):
name = ctypes.create_string_buffer(name)
elif isinstance(name, unicode):
name = ctypes.create_unicode_buffer(name)
else:
raise TypeError("`name` must be `bytes` or `unicode`")
result = _shm_unlink(name)
if result == -1:
raise RuntimeError(os.strerror(ctypes.get_errno()))
import os
import mmap
import numpy as np
from shm import *
fid = shm_open("arr")
size = 10
os.ftruncate(fid, size) # Skipped by processes reusing the memory
m = mmap.mmap(fid, size)
a = np.frombuffer(m, dtype=np.uint8)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment