refactor synchronization for simplicity

Author: alimanfoo

zarr/__init__.py

@@ -5,14 +5,13 @@
 import atexit
 
 
-from zarr.core import Array
+from zarr.core import Array, SynchronizedArray
 from zarr.creation import create, array, empty, zeros, ones, full, open, \
     empty_like, zeros_like, ones_like, full_like, open_like, open_array
 from zarr.storage import DictStore, DirectoryStore, ZipStore, init_array, \
     init_group, init_store
 from zarr.hierarchy import group, open_group, Group
-from zarr.sync import ThreadSynchronizer, ProcessSynchronizer, \
-    SynchronizedArray
+from zarr.sync import ThreadSynchronizer, ProcessSynchronizer
 from zarr.version import version as __version__
 
 

zarr/attrs.py

@@ -10,7 +10,7 @@
 
 class Attributes(MutableMapping):
 
-    def __init__(self, store, key='attrs', readonly=False):
+    def __init__(self, store, key='.zattrs', readonly=False):
         self.store = store
         self.key = key
         self.readonly = readonly
@@ -21,12 +21,7 @@ def __contains__(self, x):
     def __getitem__(self, item):
         return self.asdict()[item]
 
-    def put(self, d):
-
-        # guard conditions
-        if self.readonly:
-            raise ReadOnlyError('attributes are read-only')
-
+    def _put(self, d):
         s = json.dumps(d, indent=4, sort_keys=True, ensure_ascii=True)
         self.store[self.key] = s.encode('ascii')
 
@@ -42,8 +37,8 @@ def __setitem__(self, key, value):
         # set key value
         d[key] = value
 
-        # put modified data
-        self.put(d)
+        # _put modified data
+        self._put(d)
 
     def __delitem__(self, key):
 
@@ -57,8 +52,8 @@ def __delitem__(self, key):
         # delete key value
         del d[key]
 
-        # put modified data
-        self.put(d)
+        # _put modified data
+        self._put(d)
 
     def asdict(self):
         if self.key in self.store:
@@ -79,8 +74,8 @@ def update(self, *args, **kwargs):
         # update
         d.update(*args, **kwargs)
 
-        # put modified data
-        self.put(d)
+        # _put modified data
+        self._put(d)
 
     def __iter__(self):
         return iter(self.asdict())
@@ -96,3 +91,23 @@ def values(self):
 
     def items(self):
         return self.asdict().items()
+
+
+class SynchronizedAttributes(Attributes):
+
+    def __init__(self, store, synchronizer, key='.zattrs', readonly=False):
+        super(SynchronizedAttributes, self).__init__(store, key=key,
+                                                     readonly=readonly)
+        self.synchronizer = synchronizer
+
+    def __setitem__(self, key, value):
+        with self.synchronizer[self.key]:
+            super(SynchronizedAttributes, self).__setitem__(key, value)
+
+    def __delitem__(self, key):
+        with self.synchronizer[self.key]:
+            super(SynchronizedAttributes, self).__delitem__(key)
+
+    def update(self, *args, **kwargs):
+        with self.synchronizer[self.key]:
+            super(SynchronizedAttributes, self).update(*args, **kwargs)

zarr/core.py

@@ -13,7 +13,7 @@
     normalize_storage_path
 from zarr.storage import array_meta_key, attrs_key, listdir, getsize
 from zarr.meta import decode_array_metadata, encode_array_metadata
-from zarr.attrs import Attributes
+from zarr.attrs import Attributes, SynchronizedAttributes
 from zarr.errors import ReadOnlyError
 from zarr.compat import reduce
 
@@ -514,7 +514,7 @@ def _chunk_getitem(self, cidx, item, dest):
         try:
 
             # obtain compressed data for chunk
-            ckey = self._ckey(cidx)
+            ckey = self._chunk_key(cidx)
             cdata = self._chunk_store[ckey]
 
         except KeyError:
@@ -589,7 +589,7 @@ def _chunk_setitem(self, cidx, key, value):
             try:
 
                 # obtain compressed data for chunk
-                ckey = self._ckey(cidx)
+                ckey = self._chunk_key(cidx)
                 cdata = self._chunk_store[ckey]
 
             except KeyError:
@@ -614,10 +614,10 @@ def _chunk_setitem(self, cidx, key, value):
         cdata = self._compressor.compress(chunk)
 
         # store
-        ckey = self._ckey(cidx)
+        ckey = self._chunk_key(cidx)
         self._chunk_store[ckey] = cdata
 
-    def _ckey(self, cidx):
+    def _chunk_key(self, cidx):
         return self._key_prefix + '.'.join(map(str, cidx))
 
     def __repr__(self):
@@ -691,6 +691,13 @@ def resize(self, *args):
         if self._readonly:
             raise ReadOnlyError('array is read-only')
 
+        self._resize(*args)
+
+    def _resize(self, *args):
+
+        # N.B., private implementation to avoid need for re-entrant lock on
+        # SynchronizedArray.append().
+
         # normalize new shape argument
         old_shape = self._shape
         new_shape = normalize_resize_args(old_shape, *args)
@@ -780,7 +787,7 @@ def append(self, data, axis=0):
         )
 
         # resize
-        self.resize(new_shape)
+        self._resize(new_shape)
 
         # store data
         # noinspection PyTypeChecker
@@ -789,3 +796,88 @@ def append(self, data, axis=0):
             for i in range(len(self._shape))
         )
         self[append_selection] = data
+
+
+class SynchronizedArray(Array):
+    """Instantiate a synchronized array.
+
+    Parameters
+    ----------
+    store : MutableMapping
+        Array store, already initialized.
+    synchronizer : object
+        Array synchronizer.
+    readonly : bool, optional
+        True if array should be protected against modification.
+    path : string, optional
+        Path under which array is stored.
+    chunk_store : MutableMapping, optional
+        Separate storage for chunks. If not provided, `store` will be used
+        for storage of both chunks and metadata.
+
+    Examples
+    --------
+    >>> import zarr
+    >>> store = dict()
+    >>> zarr.init_array(store, shape=1000, chunks=100)
+    >>> synchronizer = zarr.ThreadSynchronizer()
+    >>> z = zarr.SynchronizedArray(store, synchronizer)
+    >>> z
+    zarr.sync.SynchronizedArray((1000,), float64, chunks=(100,), order=C)
+      compression: blosc; compression_opts: {'clevel': 5, 'cname': 'lz4', 'shuffle': 1}
+      nbytes: 7.8K; nbytes_stored: 285; ratio: 28.1; initialized: 0/10
+      store: builtins.dict; synchronizer: zarr.sync.ThreadSynchronizer
+
+    Notes
+    -----
+    TODO review
+
+    Only writing data to the array via the __setitem__() method and
+    modification of user attributes are synchronized. Neither append() nor
+    resize() are synchronized.
+
+    Writing to the array is synchronized at the chunk level. I.e.,
+    the array supports concurrent write operations via the __setitem__()
+    method, but these will only exclude each other if they both require
+    modification of the same chunk.
+
+    """  # flake8: noqa
+
+    def __init__(self, store, synchronizer, readonly=False, path=None,
+                 chunk_store=None):
+        super(SynchronizedArray, self).__init__(store, readonly=readonly,
+                                                path=path,
+                                                chunk_store=chunk_store)
+        self.synchronizer = synchronizer
+        akey = self._key_prefix + attrs_key
+        self._attrs = SynchronizedAttributes(store, synchronizer, key=akey,
+                                             readonly=readonly)
+
+    def __repr__(self):
+        r = super(SynchronizedArray, self).__repr__()
+        r += ('; synchronizer: %s.%s' %
+              (type(self.synchronizer).__module__,
+               type(self.synchronizer).__name__))
+        return r
+
+    def __getstate__(self):
+        return self._store, self.synchronizer, self._readonly, self._path, \
+               self._chunk_store
+
+    def __setstate__(self, state):
+        self.__init__(*state)
+
+    def _chunk_setitem(self, cidx, key, value):
+        ckey = self._chunk_key(cidx)
+        with self.synchronizer[ckey]:
+            super(SynchronizedArray, self)._chunk_setitem(cidx, key, value)
+
+    def resize(self, *args):
+        mkey = self._key_prefix + array_meta_key
+        with self.synchronizer[mkey]:
+            super(SynchronizedArray, self).resize(*args)
+
+    def append(self, *args, **kwargs):
+        mkey = self._key_prefix + array_meta_key
+        with self.synchronizer[mkey]:
+            super(SynchronizedArray, self).append(*args, **kwargs)

zarr/sync.py

@@ -18,22 +18,18 @@ class ThreadSynchronizer(object):
 
     def __init__(self):
         self.mutex = Lock()
-        self.attrs_lock = Lock()
-        self.chunk_locks = defaultdict(Lock)
+        self.locks = defaultdict(Lock)
 
-    def chunk_lock(self, ckey):
+    def __getitem__(self, item):
         with self.mutex:
-            lock = self.chunk_locks[ckey]
-        return lock
+            return self.locks[item]
 
     def __getstate__(self):
         return dict()
 
     def __setstate__(self, d):
         # reinitialize from scratch
-        self.mutex = Lock()
-        self.attrs_lock = Lock()
-        self.chunk_locks = defaultdict(Lock)
+        self.__init__()
 
 
 class ProcessSynchronizer(object):
@@ -51,104 +47,10 @@ class ProcessSynchronizer(object):
     def __init__(self, path):
         self.path = path
 
-    @property
-    def attrs_lock(self):
-        return fasteners.InterProcessLock(
-            os.path.join(self.path, 'attrs.lock')
-        )
-
-    def chunk_lock(self, ckey):
+    def __getitem__(self, item):
         lock = fasteners.InterProcessLock(
-            os.path.join(self.path, '%s.lock' % ckey)
+            os.path.join(self.path, '%s.lock' % item)
         )
         return lock
 
-
-class SynchronizedArray(Array):
-    """Instantiate a synchronized array.
-
-    Parameters
-    ----------
-    store : MutableMapping
-        Array store, already initialized.
-    synchronizer : object
-        Array synchronizer.
-    readonly : bool, optional
-        True if array should be protected against modification.
-    path : string, optional
-        Path under which array is stored.
-
-    Examples
-    --------
-    >>> import zarr
-    >>> store = dict()
-    >>> zarr.init_array(store, shape=1000, chunks=100)
-    >>> synchronizer = zarr.ThreadSynchronizer()
-    >>> z = zarr.SynchronizedArray(store, synchronizer)
-    >>> z
-    zarr.sync.SynchronizedArray((1000,), float64, chunks=(100,), order=C)
-      compression: blosc; compression_opts: {'clevel': 5, 'cname': 'lz4', 'shuffle': 1}
-      nbytes: 7.8K; nbytes_stored: 285; ratio: 28.1; initialized: 0/10
-      store: builtins.dict; synchronizer: zarr.sync.ThreadSynchronizer
-
-    Notes
-    -----
-    Only writing data to the array via the __setitem__() method and
-    modification of user attributes are synchronized. Neither append() nor
-    resize() are synchronized.
-
-    Writing to the array is synchronized at the chunk level. I.e.,
-    the array supports concurrent write operations via the __setitem__()
-    method, but these will only exclude each other if they both require
-    modification of the same chunk.
-
-    """  # flake8: noqa
-
-    def __init__(self, store, synchronizer, readonly=False, path=None,
-                 chunk_store=None):
-        super(SynchronizedArray, self).__init__(store, readonly=readonly,
-                                                path=path,
-                                                chunk_store=chunk_store)
-        self.synchronizer = synchronizer
-        akey = self._key_prefix + attrs_key
-        self._attrs = SynchronizedAttributes(store, synchronizer, key=akey,
-                                             readonly=readonly)
-
-    def _chunk_setitem(self, cidx, key, value):
-        ckey = self._ckey(cidx)
-        with self.synchronizer.chunk_lock(ckey):
-            super(SynchronizedArray, self)._chunk_setitem(cidx, key, value)
-
-    def __repr__(self):
-        r = super(SynchronizedArray, self).__repr__()
-        r += ('; synchronizer: %s.%s' %
-              (type(self.synchronizer).__module__,
-               type(self.synchronizer).__name__))
-        return r
-
-    def __getstate__(self):
-        return self._store, self.synchronizer, self._readonly, self._path, \
-               self._chunk_store
-
-    def __setstate__(self, state):
-        self.__init__(*state)
-
-
-class SynchronizedAttributes(Attributes):
-
-    def __init__(self, store, synchronizer, key='attrs', readonly=False):
-        super(SynchronizedAttributes, self).__init__(store, key=key,
-                                                     readonly=readonly)
-        self.synchronizer = synchronizer
-
-    def __setitem__(self, key, value):
-        with self.synchronizer.attrs_lock:
-            super(SynchronizedAttributes, self).__setitem__(key, value)
-
-    def __delitem__(self, key):
-        with self.synchronizer.attrs_lock:
-            super(SynchronizedAttributes, self).__delitem__(key)
-
-    def update(self, *args, **kwargs):
-        with self.synchronizer.attrs_lock:
-            super(SynchronizedAttributes, self).update(*args, **kwargs)
+    # pickling and unpickling should be handled automatically

zarr/tests/test_attrs.py

@@ -86,5 +86,3 @@ def test_readonly(self):
             del a['foo']
         with assert_raises(ReadOnlyError):
             a.update(foo='quux')
-        with assert_raises(ReadOnlyError):
-            a.put(dict())