implement group synchronization; refactor array synchronization

Author: alimanfoo

zarr/__init__.py

@@ -5,7 +5,7 @@
 import atexit
 
 
-from zarr.core import Array, SynchronizedArray
+from zarr.core import Array
 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, \

zarr/attrs.py

@@ -10,10 +10,12 @@
 
 class Attributes(MutableMapping):
 
-    def __init__(self, store, key='.zattrs', readonly=False):
+    def __init__(self, store, key='.zattrs', readonly=False,
+                 synchronizer=None):
         self.store = store
         self.key = key
         self.readonly = readonly
+        self.synchronizer = synchronizer
 
     def __contains__(self, x):
         return x in self.asdict()
@@ -25,26 +27,37 @@ def _put(self, d):
         s = json.dumps(d, indent=4, sort_keys=True, ensure_ascii=True)
         self.store[self.key] = s.encode('ascii')
 
-    def __setitem__(self, key, value):
+    def _write_op(self, f, *args, **kwargs):
 
-        # guard conditions
+        # guard condition
         if self.readonly:
             raise ReadOnlyError('attributes are read-only')
 
+        # synchronization
+        if self.synchronizer is None:
+            return f(*args, **kwargs)
+        else:
+            with self.synchronizer[self.key]:
+                return f(*args, **kwargs)
+
+    def __setitem__(self, item, value):
+        self._write_op(self._setitem_nosync, item, value)
+
+    def _setitem_nosync(self, item, value):
+
         # load existing data
         d = self.asdict()
 
         # set key value
-        d[key] = value
+        d[item] = value
 
         # _put modified data
         self._put(d)
 
-    def __delitem__(self, key):
+    def __delitem__(self, item):
+        self._write_op(self._delitem_nosync, item)
 
-        # guard conditions
-        if self.readonly:
-            raise ReadOnlyError('attributes are read-only')
+    def _delitem_nosync(self, key):
 
         # load existing data
         d = self.asdict()
@@ -63,10 +76,9 @@ def asdict(self):
 
     def update(self, *args, **kwargs):
         # override to provide update in a single write
+        self._write_op(self._update_nosync, *args, **kwargs)
 
-        # guard conditions
-        if self.readonly:
-            raise ReadOnlyError('mapping is read-only')
+    def _update_nosync(self, *args, **kwargs):
 
         # load existing data
         d = self.asdict()
@@ -82,32 +94,3 @@ def __iter__(self):
 
     def __len__(self):
         return len(self.asdict())
-
-    def keys(self):
-        return self.asdict().keys()
-
-    def values(self):
-        return self.asdict().values()
-
-    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, SynchronizedAttributes
+from zarr.attrs import Attributes
 from zarr.errors import ReadOnlyError
 from zarr.compat import reduce
 
@@ -32,6 +32,8 @@ class Array(object):
     chunk_store : MutableMapping, optional
         Separate storage for chunks. If not provided, `store` will be used
         for storage of both chunks and metadata.
+    synchronizer : object, optional
+        Array synchronizer.
 
     Attributes
     ----------
@@ -47,6 +49,7 @@ class Array(object):
     compression_opts
     fill_value
     order
+    synchronizer
     attrs
     size
     itemsize
@@ -64,7 +67,8 @@ class Array(object):
 
     """  # flake8: noqa
 
-    def __init__(self, store, path=None, readonly=False, chunk_store=None):
+    def __init__(self, store, path=None, readonly=False, chunk_store=None,
+                 synchronizer=None):
         # N.B., expect at this point store is fully initialized with all
         # configuration metadata fully specified and normalized
 
@@ -79,6 +83,7 @@ def __init__(self, store, path=None, readonly=False, chunk_store=None):
             self._chunk_store = store
         else:
             self._chunk_store = chunk_store
+        self._synchronizer = synchronizer
 
         # initialize metadata
         try:
@@ -101,9 +106,10 @@ def __init__(self, store, path=None, readonly=False, chunk_store=None):
 
         # initialize attributes
         akey = self._key_prefix + attrs_key
-        self._attrs = Attributes(store, key=akey, readonly=readonly)
+        self._attrs = Attributes(store, key=akey, readonly=readonly,
+                                 synchronizer=synchronizer)
 
-    def flush_metadata(self):
+    def _flush_metadata(self):
         meta = dict(shape=self._shape, chunks=self._chunks, dtype=self._dtype,
                     compression=self._compression,
                     compression_opts=self._compression_opts,
@@ -183,6 +189,11 @@ def order(self):
         chunks of the array."""
         return self._order
 
+    @property
+    def synchronizer(self):
+        """TODO doc me"""
+        return self._synchronizer
+
     @property
     def attrs(self):
         """A MutableMapping containing user-defined attributes. Note that
@@ -563,6 +574,20 @@ def _chunk_setitem(self, cidx, key, value):
 
         """
 
+        # synchronization
+        if self._synchronizer is None:
+            self._chunk_setitem_nosync(cidx, key, value)
+        else:
+            # synchronize on the chunk
+            ckey = self._chunk_key(cidx)
+            with self._synchronizer[ckey]:
+                self._chunk_setitem_nosync(cidx, key, value)
+
+    def _chunk_setitem_nosync(self, cidx, key, value):
+
+        # obtain key for chunk storage
+        ckey = self._chunk_key(cidx)
+
         if is_total_slice(key, self._chunks):
 
             # optimisation: we are completely replacing the chunk, so no need
@@ -589,7 +614,6 @@ def _chunk_setitem(self, cidx, key, value):
             try:
 
                 # obtain compressed data for chunk
-                ckey = self._chunk_key(cidx)
                 cdata = self._chunk_store[ckey]
 
             except KeyError:
@@ -614,7 +638,6 @@ def _chunk_setitem(self, cidx, key, value):
         cdata = self._compressor.compress(chunk)
 
         # store
-        ckey = self._chunk_key(cidx)
         self._chunk_store[ckey] = cdata
 
     def _chunk_key(self, cidx):
@@ -644,14 +667,34 @@ def __repr__(self):
             r += '\n  chunk_store: %s.%s' % \
                  (type(self._chunk_store).__module__,
                   type(self._chunk_store).__name__)
+        if self._synchronizer is not None:
+            r += ('\n  synchronizer: %s.%s' %
+                  (type(self._synchronizer).__module__,
+                   type(self._synchronizer).__name__))
         return r
 
     def __getstate__(self):
-        return self._store, self._path, self._readonly, self._chunk_store
+        return self._store, self._path, self._readonly, self._chunk_store, \
+               self._synchronizer
 
     def __setstate__(self, state):
         self.__init__(*state)
 
+    def _write_op(self, f, *args, **kwargs):
+
+        # guard condition
+        if self._readonly:
+            raise ReadOnlyError('array is read-only')
+
+        # synchronization
+        if self._synchronizer is None:
+            return f(*args, **kwargs)
+        else:
+            # synchronize on the array
+            mkey = self._key_prefix + array_meta_key
+            with self._synchronizer[mkey]:
+                return f(*args, **kwargs)
+
     def resize(self, *args):
         """Change the shape of the array by growing or shrinking one or more
         dimensions.
@@ -687,16 +730,9 @@ def resize(self, *args):
 
         """  # flake8: noqa
 
-        # guard conditions
-        if self._readonly:
-            raise ReadOnlyError('array is read-only')
-
-        self._resize(*args)
+        return self._write_op(self._resize_nosync, *args)
 
-    def _resize(self, *args):
-
-        # N.B., private implementation to avoid need for re-entrant lock on
-        # SynchronizedArray.append().
+    def _resize_nosync(self, *args):
 
         # normalize new shape argument
         old_shape = self._shape
@@ -718,7 +754,7 @@ def _resize(self, *args):
 
         # update metadata
         self._shape = new_shape
-        self.flush_metadata()
+        self._flush_metadata()
 
     def append(self, data, axis=0):
         """Append `data` to `axis`.
@@ -760,10 +796,9 @@ def append(self, data, axis=0):
           store: builtins.dict
 
         """
+        return self._write_op(self._append_nosync, data, axis=axis)
 
-        # guard conditions
-        if self._readonly:
-            raise ReadOnlyError('array is read-only')
+    def _append_nosync(self, data, axis=0):
 
         # ensure data is array-like
         if not hasattr(data, 'shape') or not hasattr(data, 'dtype'):
@@ -787,7 +822,7 @@ def append(self, data, axis=0):
         )
 
         # resize
-        self._resize(new_shape)
+        self._resize_nosync(new_shape)
 
         # store data
         # noinspection PyTypeChecker
@@ -796,88 +831,3 @@ 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.core.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 += ('\n  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)