Some Python ctypes-based POSIX shared memory functions

shm.py

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

shm_example.py

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)