fix nasty process sync bug

Author: alimanfoo

zarr/core.py

@@ -148,7 +148,7 @@ def _refresh_metadata(self):
             self._load_metadata()
 
     def _refresh_metadata_nosync(self):
-        if not self._cache_metadata:
+        if not self._cache_metadata and not self._is_view:
             self._load_metadata_nosync()
 
     def _flush_metadata_nosync(self):
@@ -205,6 +205,8 @@ def chunk_store(self):
     def shape(self):
         """A tuple of integers describing the length of each dimension of
         the array."""
+        # N.B., shape may change if array is resized, hence need to refresh
+        # metadata
         self._refresh_metadata()
         return self._shape
 
@@ -267,6 +269,8 @@ def _size(self):
     @property
     def size(self):
         """The total number of elements in the array."""
+        # N.B., this property depends on shape, and shape may change if array
+        # is resized, hence need to refresh metadata
         self._refresh_metadata()
         return self._size
 
@@ -283,6 +287,8 @@ def _nbytes(self):
     def nbytes(self):
         """The total number of bytes that would be required to store the
         array without compression."""
+        # N.B., this property depends on shape, and shape may change if array
+        # is resized, hence need to refresh metadata
         self._refresh_metadata()
         return self._nbytes
 
@@ -559,7 +565,7 @@ def __setitem__(self, key, value):
 
         # refresh metadata
         if not self._cache_metadata:
-            self._load_metadata()
+            self._load_metadata_nosync()
 
         # normalize selection
         selection = normalize_array_selection(key, self._shape)
@@ -793,7 +799,11 @@ def _encode_chunk(self, chunk):
         return cdata
 
     def __repr__(self):
-        self._refresh_metadata()
+        # N.B., __repr__ needs to be synchronized to ensure consistent view
+        # of metadata AND when retrieving nbytes_stored from filesystem storage
+        return self._synchronized_op(self._repr_nosync)
+
+    def _repr_nosync(self):
 
         # main line
         r = '%s(' % type(self).__name__
@@ -842,13 +852,9 @@ def __getstate__(self):
     def __setstate__(self, state):
         self.__init__(*state)
 
-    def _write_op(self, f, *args, **kwargs):
-
-        # guard condition
-        if self._read_only:
-            err_read_only()
+    def _synchronized_op(self, f, *args, **kwargs):
 
-        # synchronization
+        # no synchronization
         if self._synchronizer is None:
             self._refresh_metadata_nosync()
             return f(*args, **kwargs)
@@ -858,7 +864,16 @@ def _write_op(self, f, *args, **kwargs):
             mkey = self._key_prefix + array_meta_key
             with self._synchronizer[mkey]:
                 self._refresh_metadata_nosync()
-                return f(*args, **kwargs)
+                result = f(*args, **kwargs)
+            return result
+
+    def _write_op(self, f, *args, **kwargs):
+
+        # guard condition
+        if self._read_only:
+            err_read_only()
+
+        return self._synchronized_op(f, *args, **kwargs)
 
     def resize(self, *args):
         """Change the shape of the array by growing or shrinking one or more
@@ -915,11 +930,15 @@ def _resize_nosync(self, *args):
         # remove any chunks not within range
         for key in listdir(self._chunk_store, self._path):
             if key not in [array_meta_key, attrs_key]:
-                cidx = map(int, key.split('.'))
-                if all(i < c for i, c in zip(cidx, new_cdata_shape)):
-                    pass  # keep the chunk
+                try:
+                    cidx = list(map(int, key.split('.')))
+                except ValueError as e:
+                    raise RuntimeError('unexpected key: %r' % key)
                 else:
-                    del self._chunk_store[self._key_prefix + key]
+                    if all(i < c for i, c in zip(cidx, new_cdata_shape)):
+                        pass  # keep the chunk
+                    else:
+                        del self._chunk_store[self._key_prefix + key]
 
     def append(self, data, axis=0):
         """Append `data` to `axis`.

zarr/storage.py

@@ -17,9 +17,9 @@
 from zarr.meta import encode_array_metadata, encode_group_metadata
 from zarr.compat import PY2, binary_type
 from zarr.codecs import codec_registry
-from zarr.errors import PermissionError, err_contains_group, \
-    err_contains_array, err_path_not_found, err_bad_compressor, \
-    err_fspath_exists_notdir, err_read_only
+from zarr.errors import err_contains_group, err_contains_array, \
+    err_path_not_found, err_bad_compressor, err_fspath_exists_notdir, \
+    err_read_only
 
 
 array_meta_key = '.zarray'

zarr/sync.py

@@ -5,6 +5,9 @@
 import os
 
 
+import fasteners
+
+
 class ThreadSynchronizer(object):
     """Provides synchronization using thread locks."""
 
@@ -41,10 +44,8 @@ def __init__(self, path):
         self.path = path
 
     def __getitem__(self, item):
-        import fasteners
-        lock = fasteners.InterProcessLock(
-            os.path.join(self.path, item)
-        )
+        path = os.path.join(self.path, item)
+        lock = fasteners.InterProcessLock(path)
         return lock
 
     # pickling and unpickling should be handled automatically

zarr/tests/test_sync.py

@@ -4,10 +4,12 @@
 import atexit
 import json
 import shutil
-import os
-from multiprocessing.pool import ThreadPool, Pool as ProcessPool
+from multiprocessing.pool import ThreadPool
+from multiprocessing import Pool as ProcessPool
 from multiprocessing import cpu_count
 import tempfile
+import traceback
+import sys
 
 
 import numpy as np
@@ -50,20 +52,28 @@ def init_attributes(self, store, read_only=False):
 
 
 def _append(arg):
-    z, i = arg
-    import numpy as np
-    x = np.empty(1000, dtype='i4')
-    x[:] = i
-    shape = z.append(x)
-    return shape
+    try:
+        z, i = arg
+        import numpy as np
+        x = np.empty(1000, dtype='i4')
+        x[:] = i
+        shape = z.append(x)
+        return shape
+    except Exception as e:
+        traceback.print_exc(file=sys.stderr)
+        raise
 
 
 def _set_arange(arg):
-    z, i = arg
-    import numpy as np
-    x = np.arange(i*1000, (i*1000)+1000, 1)
-    z[i*1000:(i*1000)+1000] = x
-    return i
+    try:
+        z, i = arg
+        import numpy as np
+        x = np.arange(i*1000, (i*1000)+1000, 1)
+        z[i*1000:(i*1000)+1000] = x
+        return i
+    except Exception as e:
+        traceback.print_exc(file=sys.stderr)
+        raise
 
 
 class MixinArraySyncTests(object):
@@ -77,15 +87,15 @@ def test_parallel_setitem(self):
         pool = self.create_pool()
 
         # parallel setitem
-        future = pool.map_async(_set_arange, zip([arr] * n, range(n)))
-        results = sorted(future.get(60))
-        pool.close()
-        pool.terminate()
+        results = pool.map(_set_arange, zip([arr] * n, range(n)), chunksize=1)
+        results = sorted(results)
 
         print(results)
         eq(list(range(n)), results)
         assert_array_equal(np.arange(n * 1000), arr[:])
 
+        pool.terminate()
+
     def test_parallel_append(self):
         n = 100
 
@@ -95,15 +105,15 @@ def test_parallel_append(self):
         pool = self.create_pool()
 
         # parallel append
-        future = pool.map_async(_append, zip([arr] * n, range(n)))
-        results = sorted(future.get(60))
-        pool.close()
-        pool.terminate()
+        results = pool.map(_append, zip([arr] * n, range(n)), chunksize=1)
+        results = sorted(results)
 
         print(results)
         eq([((i+2)*1000,) for i in range(n)], results)
         eq(((n+1)*1000,), arr.shape)
 
+        pool.terminate()
+
 
 class TestArrayWithThreadSynchronizer(TestArray, MixinArraySyncTests):
 
@@ -136,12 +146,13 @@ def create_pool(self):
 class TestArrayWithProcessSynchronizer(TestArray, MixinArraySyncTests):
 
     def create_array(self, read_only=False, **kwargs):
-        path = 'test_sync'
-        if os.path.exists(path):
-            shutil.rmtree(path)
+        path = tempfile.mkdtemp()
+        atexit.register(atexit_rmtree, path)
         store = DirectoryStore(path)
         init_array(store, **kwargs)
-        synchronizer = ProcessSynchronizer('test_sync_locks')
+        sync_path = tempfile.mkdtemp()
+        atexit.register(atexit_rmtree, sync_path)
+        synchronizer = ProcessSynchronizer(sync_path)
         return Array(store, synchronizer=synchronizer,
                      read_only=read_only, cache_metadata=False)
 
@@ -161,7 +172,7 @@ def test_repr(self):
                 eq(l1, l2)
 
     def create_pool(self):
-        pool = ProcessPool(cpu_count())
+        pool = ProcessPool(processes=cpu_count())
         return pool
 
 
@@ -187,15 +198,20 @@ def test_parallel_create_group(self):
 
         # parallel create group
         n = 100
-        future = pool.map_async(
-            _create_group, zip([g] * n, [str(i) for i in range(n)]))
-        results = sorted(future.get(60))
+        results = pool.map(
+            _create_group,
+            zip([g] * n, [str(i) for i in range(n)]),
+            chunksize=1
+        )
+        results = sorted(results)
         pool.close()
         pool.terminate()
 
         print(results)
         eq(n, len(g))
 
+        pool.terminate()
+
     def test_parallel_require_group(self):
 
         # setup
@@ -204,15 +220,20 @@ def test_parallel_require_group(self):
 
         # parallel require group
         n = 100
-        future = pool.map_async(
-            _require_group, zip([g] * n, [str(i//10) for i in range(n)]))
-        results = sorted(future.get(60))
+        results = pool.map(
+            _require_group,
+            zip([g] * n, [str(i//10) for i in range(n)]),
+            chunksize=1
+        )
+        results = sorted(results)
         pool.close()
         pool.terminate()
 
         print(results)
         eq(n//10, len(g))
 
+        pool.terminate()
+
 
 class TestGroupWithThreadSynchronizer(TestGroup, MixinGroupSyncTests):
 
@@ -247,9 +268,8 @@ def test_synchronizer_property(self):
 class TestGroupWithProcessSynchronizer(TestGroup, MixinGroupSyncTests):
 
     def create_store(self):
-        path = 'test_sync'
-        if os.path.exists(path):
-            shutil.rmtree(path)
+        path = tempfile.mkdtemp()
+        atexit.register(atexit_rmtree, path)
         store = DirectoryStore(path)
         return store, None
 
@@ -258,13 +278,15 @@ def create_group(self, store=None, path=None, read_only=False,
         if store is None:
             store, chunk_store = self.create_store()
         init_group(store, path=path, chunk_store=chunk_store)
-        synchronizer = ProcessSynchronizer('test_sync_locks')
+        sync_path = tempfile.mkdtemp()
+        atexit.register(atexit_rmtree, sync_path)
+        synchronizer = ProcessSynchronizer(sync_path)
         g = Group(store, path=path, read_only=read_only,
                   synchronizer=synchronizer, chunk_store=chunk_store)
         return g
 
     def create_pool(self):
-        pool = ProcessPool(cpu_count())
+        pool = ProcessPool(processes=cpu_count())
         return pool
 
     def test_group_repr(self):