add synchronization for persistent arrays

Author: alimanfoo

README.rst

@@ -123,7 +123,7 @@ Create a persistent array (data saved to disk)::
     >>> z = zarr.open(path, shape=(10000, 1000), dtype='i4', chunks=(1000, 100))
     >>> z[:] = np.arange(10000000, dtype='i4').reshape(10000, 1000)
     >>> z
-    zarr.ext.PersistentArray((10000, 1000), int32, chunks=(1000, 100))
+    zarr.ext.SynchronizedPersistentArray((10000, 1000), int32, chunks=(1000, 100))
       cname: blosclz; clevel: 5; shuffle: 1 (BYTESHUFFLE)
       nbytes: 38.1M; cbytes: 2.0M; ratio: 19.3; initialized: 100/100
       mode: a; path: example.zarr

setup.py

@@ -74,6 +74,9 @@
         'setuptools>18.0',
         'setuptools-scm>1.5.4'
     ],
+    install_requires=[
+        'fasteners',
+    ],
     ext_modules=ext_modules,
     package_dir={'': '.'},
     packages=['zarr', 'zarr.tests'],

zarr/core.py

@@ -166,7 +166,7 @@ def full(shape, chunks, fill_value, dtype=None, cname=None, clevel=None,
 
 
 def array(data, chunks=None, dtype=None, cname=None, clevel=None,
-          shuffle=None, synchronized=True, fill_value=None):
+          shuffle=None, fill_value=None, synchronized=True):
     """Create an array filled with `data`.
 
     Parameters
@@ -185,11 +185,11 @@ def array(data, chunks=None, dtype=None, cname=None, clevel=None,
     shuffle : int, optional
         Shuffle filter, 0 means no shuffle, 1 means byte shuffle, 2 means
         bit shuffle.
+    fill_value : object
+        Default value to use for uninitialised portions of the array.
     synchronized : bool, optional
         If True, each chunk will be protected with a lock to prevent data
         collision during write operations.
-    fill_value : object
-        Default value to use for uninitialised portions of the array.
 
     Returns
     -------
@@ -233,8 +233,9 @@ def array(data, chunks=None, dtype=None, cname=None, clevel=None,
     return z
 
 
+# noinspection PyShadowingBuiltins
 def open(path, mode='a', shape=None, chunks=None, dtype=None, cname=None,
-         clevel=None, shuffle=None, fill_value=None):
+         clevel=None, shuffle=None, fill_value=None, synchronized=True):
     """Open a persistent array.
 
     Parameters
@@ -262,16 +263,20 @@ def open(path, mode='a', shape=None, chunks=None, dtype=None, cname=None,
         bit shuffle.
     fill_value : object
         Default value to use for uninitialised portions of the array.
+    synchronized : bool, optional
+        If True, each chunk will be protected with a lock to prevent data
+        collision during write operations.
 
     Returns
     -------
     z : zarr Array
 
     """
 
-    # TODO synchronized option
-
-    cls = _ext.PersistentArray
+    if synchronized:
+        cls = _ext.SynchronizedPersistentArray
+    else:
+        cls = _ext.PersistentArray
     return cls(path=path, mode=mode, shape=shape, chunks=chunks, dtype=dtype,
                cname=cname, clevel=clevel, shuffle=shuffle,
                fill_value=fill_value)

zarr/ext.pxd

@@ -37,8 +37,8 @@ cdef class PersistentChunk(BaseChunk):
 
 
 cdef class SynchronizedPersistentChunk(PersistentChunk):
-    # TODO
-    pass
+    cdef object _thread_lock
+    cdef object _file_lock
 
 
 cdef class BaseArray:
@@ -73,7 +73,6 @@ cdef class PersistentArray(BaseArray):
 
 
 cdef class SynchronizedPersistentArray(PersistentArray):
-    # TODO
     pass
 
 

zarr/ext.pyx

@@ -19,6 +19,7 @@ import tempfile
 from collections import namedtuple
 
 
+import fasteners
 from zarr import util as _util
 from zarr import defaults
 
@@ -574,13 +575,15 @@ cdef class PersistentChunk(BaseChunk):
 # noinspection PyAbstractClass
 cdef class SynchronizedPersistentChunk(PersistentChunk):
 
-    def __cinit__(self, **kargs):
-        # TODO
-        pass
+    def __cinit__(self, **kwargs):
+        lock_path = self._path + '.lock'
+        self._thread_lock = RLock()
+        self._file_lock = fasteners.InterProcessLock(lock_path)
 
     def __setitem__(self, key, value):
-        # TODO
-        pass
+        with self._thread_lock:
+            with self._file_lock:
+                super(SynchronizedPersistentChunk, self).__setitem__(key, value)
 
 
 ###############################################################################
@@ -1303,9 +1306,18 @@ cdef class PersistentArray(BaseArray):
             yield self._cdata[cidx]
 
 
+# noinspection PyAbstractClass
 cdef class SynchronizedPersistentArray(PersistentArray):
-    # TODO
-    pass
+
+    cdef BaseChunk create_chunk(self, tuple cidx):
+        chunk_filename = '.'.join(map(str, cidx)) + defaults.datasuffix
+        chunk_path = os.path.join(self._path, defaults.datapath,
+                                  chunk_filename)
+        return SynchronizedPersistentChunk(
+            path=chunk_path, shape=self._chunks, dtype=self._dtype,
+            cname=self._cname, clevel=self._clevel, shuffle=self._shuffle,
+            fill_value=self._fill_value
+        )
 
 
 cdef class LazyArray(BaseArray):

zarr/tests/test_array.py

@@ -10,7 +10,8 @@
 from nose.tools import eq_ as eq, assert_false, assert_true, assert_raises
 import numpy as np
 from numpy.testing import assert_array_equal
-from zarr.ext import Array, SynchronizedArray, PersistentArray
+from zarr.ext import Array, SynchronizedArray, PersistentArray, \
+    SynchronizedPersistentArray
 from zarr import defaults
 
 
@@ -363,3 +364,15 @@ def test_persistence_2d(self):
         a = np.arange(10000).reshape((1000, 10))
         chunks = (100, 2)
         self._test_persistence(a, chunks=chunks)
+
+
+class TestSynchronizedPersistentArray(TestPersistentArray):
+
+    def create_array(self, **kwargs):
+        path = kwargs.get('path', tempfile.mktemp())
+        kwargs['path'] = path
+        # tidy up
+        atexit.register(
+            lambda: shutil.rmtree(path) if os.path.exists(path) else None
+        )
+        return SynchronizedPersistentArray(**kwargs)

zarr/tests/test_chunk.py

@@ -9,7 +9,8 @@
 from nose.tools import eq_ as eq, assert_false, assert_true, assert_raises
 import numpy as np
 from numpy.testing import assert_array_equal
-from zarr.ext import Chunk, PersistentChunk, SynchronizedChunk
+from zarr.ext import Chunk, PersistentChunk, SynchronizedChunk, \
+    SynchronizedPersistentChunk
 from zarr import defaults
 
 
@@ -223,4 +224,13 @@ def test_persistence(self):
                                    .reshape(100, -1))
 
 
-# TODO test persistent synchronized chunks
+class TestSynchronizedPersistentChunk(TestPersistentChunk):
+
+    def create_chunk(self, **kwargs):
+        path = kwargs.get('path', tempfile.mktemp())
+        kwargs['path'] = path
+        # tidy up
+        atexit.register(
+            lambda: os.remove(path) if os.path.exists(path) else None
+        )
+        return SynchronizedPersistentChunk(**kwargs)