Create a trivial char array object

This builds up a basic char array object in C with some Python trappings
for handling construction and cleanup of the object. Otherwise provides
no interface for accessing this object in Python. Since it's purpose is
merely to manage a block of memory requested by the user to be passed
off and used by other objects that implement the buffer protocol, there
is no need for it to have other functionality. It simply allocates
memory from Python's memory allocator and frees it on cleanup.
Implements the buffer protocol in Cython for this object. Thus allowing
the memory allocated to be reused by NumPy arrays or other Python
objects that support the buffer protocol.

setup.py

@@ -70,6 +70,7 @@
 extra_compile_args = []
 cython_directives = {}
 cython_line_directives = {}
+cython_compile_time_env = {"PY_VERSION_HEX": sys.hexversion}
 
 
 if "test" in sys.argv:
@@ -99,6 +100,7 @@
 for em in ext_modules:
     em.cython_directives = dict(cython_directives)
     em.cython_line_directives = dict(cython_line_directives)
+    em.cython_compile_time_env = dict(cython_compile_time_env)
 
 
 setup(

src/pysharedmem.pyx

@@ -1,3 +1,76 @@
+cimport libc
+cimport libc.string
+from libc.string cimport memcpy
+
+cimport cython
+
 cimport pysharedmem
 
 include "version.pxi"
+
+
+cdef extern from "Python.h":
+    IF PY_VERSION_HEX >= 0x03040000:
+        void* PyMem_RawMalloc(size_t n) nogil
+        void PyMem_RawFree(void *p) nogil
+    ELSE:
+        void* PyMem_RawMalloc "PyMem_Malloc" (size_t n) nogil
+        void PyMem_RawFree "PyMem_Free" (void *p) nogil
+
+
+
+cdef class cbuffer:
+    cdef char[::1] buf
+
+
+    def __cinit__(self, Py_ssize_t length):
+        self.buf = None
+
+        if length <= 0:
+            raise ValueError("length must be positive definite")
+
+        cdef void* ptr = PyMem_RawMalloc(length * sizeof(char))
+        if not ptr:
+            raise MemoryError("unable to allocate buffer")
+
+        self.buf = <char[:length]>ptr
+
+
+    def __getbuffer__(self, Py_buffer *buffer, int flags):
+        buffer.buf = &self.buf[0]
+        buffer.obj = self
+        buffer.len = len(self.buf)
+        buffer.readonly = 0
+        buffer.itemsize = self.buf.ndim
+        buffer.format = "c"
+        buffer.ndim = self.buf.ndim
+        buffer.shape = self.buf.shape
+        buffer.strides = self.buf.strides
+        buffer.suboffsets = NULL
+        buffer.internal = NULL
+
+
+    def __releasebuffer__(self, Py_buffer *buffer):
+        pass
+
+
+    def __reduce__(self):
+        return (__create_cbuffer_data, (bytearray(memoryview(self)),))
+
+
+    @cython.nonecheck(False)
+    def __dealloc__(self):
+        if self.buf is None:
+            PyMem_RawFree(&self.buf[0])
+            self.buf = None
+
+
+def __create_cbuffer_data(bytearray data):
+    cdef char[:] data_buf = data
+    cdef Py_ssize_t data_len = len(data_buf)
+
+    cdef cbuffer new = cbuffer(data_len)
+
+    memcpy(&new.buf[0], &data_buf[0], data_len)
+
+    return new

tests/test_pysharedmem.py

@@ -3,12 +3,125 @@
 
 from __future__ import absolute_import
 
+import pickle
+import sys
 import unittest
 
 
-class TestImportTopLevel(unittest.TestCase):
-    def runTest(self):
+try:
+    unichr
+except NameError:
+    unichr = chr
+
+
+class TestPySharedMem(unittest.TestCase):
+    def test_import(self):
         try:
             import pysharedmem
         except ImportError:
             self.fail("Unable to import `pysharedmem`.")
+
+
+    def test_create_empty_mem(self):
+        import pysharedmem
+        with self.assertRaises(ValueError):
+            pysharedmem.cbuffer(0)
+
+
+    def test_create_negative_mem(self):
+        import pysharedmem
+        with self.assertRaises(ValueError):
+            pysharedmem.cbuffer(-1)
+
+
+    def test_create_singleton_mem(self):
+        import pysharedmem
+        pysharedmem.cbuffer(1)
+
+
+    def test_memoryview(self):
+        import pysharedmem
+
+        l = 3
+        b = pysharedmem.cbuffer(l)
+
+        m = memoryview(b)
+
+        self.assertEqual(m.readonly, False)
+
+        self.assertEqual(m.format, 'c')
+        self.assertEqual(m.itemsize, 1)
+
+        self.assertEqual(m.ndim, 1)
+        self.assertEqual(m.shape, (l,))
+        self.assertEqual(m.strides, (1,))
+
+        if sys.version_info[0] >= 3:
+            self.assertEqual(m.suboffsets, ())
+        else:
+            self.assertEqual(m.suboffsets, None)
+
+
+    @unittest.skipIf(sys.version_info[0] < 3,
+                     "This test requires Python 3.")
+    def test_memoryview_py3(self):
+        import pysharedmem
+
+        l = 3
+        b = pysharedmem.cbuffer(l)
+
+        m = memoryview(b)
+
+        self.assertIs(m.obj, b)
+
+        self.assertTrue(m.c_contiguous)
+        self.assertTrue(m.contiguous)
+        self.assertTrue(m.f_contiguous)
+
+        self.assertEqual(m.readonly, False)
+
+        self.assertEqual(m.format, 'c')
+        self.assertEqual(m.itemsize, 1)
+
+        self.assertEqual(m.ndim, 1)
+        self.assertEqual(m.nbytes, l)
+        self.assertEqual(m.shape, (l,))
+        self.assertEqual(m.strides, (1,))
+        self.assertEqual(m.suboffsets, ())
+
+
+    def test_assignment(self):
+        import pysharedmem
+
+        l = 3
+        b = pysharedmem.cbuffer(l)
+
+        m = memoryview(b)
+        for i in range(len(m)):
+            m[i] = unichr(i).encode("utf-8")
+
+        a1 = bytearray(m)
+        a2 = bytearray(b)
+
+        self.assertIsNot(a1, a2)
+        self.assertEqual(a1, a2)
+
+
+    def test_pickle(self):
+        import pysharedmem
+
+        l = 3
+        b1 = pysharedmem.cbuffer(l)
+
+        m = memoryview(b1)
+        for i in range(len(m)):
+            m[i] = unichr(i).encode("utf-8")
+
+        b2 = pickle.loads(pickle.dumps(b1))
+
+        a1 = bytes(memoryview(b1))
+        a2 = bytes(memoryview(b2))
+
+        self.assertIsNot(b1, b2)
+        self.assertIsNot(a1, a2)
+        self.assertEqual(a1, a2)