Skip to content

Instantly share code, notes, and snippets.

@jakirkham
Last active January 20, 2024 15:27
Show Gist options
  • Star 14 You must be signed in to star a gist
  • Fork 3 You must be signed in to fork a gist
  • Save jakirkham/100a7f5e86b0ff2a22de0850723a4c5c to your computer and use it in GitHub Desktop.
Save jakirkham/100a7f5e86b0ff2a22de0850723a4c5c 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)
@mara004
Copy link

mara004 commented Jul 16, 2023

The string buffer handling here is slightly confusing. Can't we directly pass in an encoded name (i.e. bytes), at least if we set argtypes for the functions?

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment