parallel tests

Author: alimanfoo

docs/release.rst

@@ -7,10 +7,9 @@ Release notes
 -----
 
 Various minor improvements, including: ``Group`` objects support member access
-via dot notation (``__getattr__``); fix metadata caching for ``Array.shape``
-property and derivatives; add ``Array.ndim`` property; fix ``Array.__array__``
-method arguments;
-
+via dot notation (``__getattr__``); fixed metadata caching for ``Array.shape``
+property and derivatives; added ``Array.ndim`` property; fixed
+``Array.__array__`` method arguments; fixed bug in pickling ``Array`` state.
 
 .. _release_2.1.0:
 

zarr/storage.py

@@ -61,7 +61,7 @@ def contains_group(store, path=None):
 def _rmdir_from_keys(store, path=None):
     # assume path already normalized
     prefix = _path_to_prefix(path)
-    for key in set(store.keys()):
+    for key in list(store.keys()):
         if key.startswith(prefix):
             del store[key]
 
@@ -81,7 +81,7 @@ def _listdir_from_keys(store, path=None):
     # assume path already normalized
     prefix = _path_to_prefix(path)
     children = set()
-    for key in store.keys():
+    for key in list(store.keys()):
         if key.startswith(prefix) and len(key) > len(prefix):
             suffix = key[len(prefix):]
             child = suffix.split('/')[0]
@@ -718,9 +718,9 @@ def getsize(self, path=None):
             err_path_not_found(path)
 
 
-def _atexit_rmtree(path,
-                   isdir=os.path.isdir,
-                   rmtree=shutil.rmtree):  # pragma: no cover
+def atexit_rmtree(path,
+                  isdir=os.path.isdir,
+                  rmtree=shutil.rmtree):  # pragma: no cover
     """Ensure directory removal at interpreter exit."""
     if isdir(path):
         rmtree(path)
@@ -731,7 +731,7 @@ class TempStore(DirectoryStore):
 
     def __init__(self, suffix='', prefix='zarr', dir=None):
         path = tempfile.mkdtemp(suffix=suffix, prefix=prefix, dir=dir)
-        atexit.register(_atexit_rmtree, path)
+        atexit.register(atexit_rmtree, path)
         super(TempStore, self).__init__(path)
 
 

zarr/tests/test_hierarchy.py

@@ -22,7 +22,6 @@
 from zarr.errors import PermissionError
 from zarr.creation import open_array
 from zarr.compat import PY2
-from zarr.sync import ThreadSynchronizer, ProcessSynchronizer
 from zarr.codecs import Zlib
 
 
@@ -752,60 +751,6 @@ def test_chunk_store(self):
         eq(expect, actual)
 
 
-class TestGroupWithThreadSynchronizer(TestGroup):
-
-    def create_group(self, store=None, path=None, read_only=False,
-                     chunk_store=None, synchronizer=None):
-        if store is None:
-            store, chunk_store = self.create_store()
-        init_group(store, path=path, chunk_store=chunk_store)
-        synchronizer = ThreadSynchronizer()
-        g = Group(store, path=path, read_only=read_only,
-                  chunk_store=chunk_store, synchronizer=synchronizer)
-        return g
-
-    def test_group_repr(self):
-        if not PY2:
-            g = self.create_group()
-            expect = 'Group(/, 0)\n' \
-                     '  store: dict; synchronizer: ThreadSynchronizer'
-            actual = repr(g)
-            for l1, l2 in zip(expect.split('\n'), actual.split('\n')):
-                eq(l1, l2)
-
-    def test_synchronizer_property(self):
-        g = self.create_group()
-        assert_is_instance(g.synchronizer, ThreadSynchronizer)
-
-
-class TestGroupWithProcessSynchronizer(TestGroup):
-
-    def create_group(self, store=None, path=None, read_only=False,
-                     chunk_store=None, synchronizer=None):
-        if store is None:
-            store, chunk_store = self.create_store()
-        init_group(store, path=path, chunk_store=chunk_store)
-        sync_path = tempfile.mkdtemp()
-        atexit.register(shutil.rmtree, sync_path)
-        synchronizer = ProcessSynchronizer(sync_path)
-        g = Group(store, path=path, read_only=read_only,
-                  chunk_store=chunk_store, synchronizer=synchronizer)
-        return g
-
-    def test_group_repr(self):
-        if not PY2:
-            g = self.create_group()
-            expect = 'Group(/, 0)\n' \
-                     '  store: dict; synchronizer: ProcessSynchronizer'
-            actual = repr(g)
-            for l1, l2 in zip(expect.split('\n'), actual.split('\n')):
-                eq(l1, l2)
-
-    def test_synchronizer_property(self):
-        g = self.create_group()
-        assert_is_instance(g.synchronizer, ProcessSynchronizer)
-
-
 def test_group():
     # test the group() convenience function
 

zarr/tests/test_sync.py

@@ -6,25 +6,38 @@
 import shutil
 from multiprocessing.pool import ThreadPool, Pool as ProcessPool
 from multiprocessing import cpu_count
-import os
 import tempfile
 
 
 import numpy as np
-from nose.tools import eq_ as eq
+from nose.tools import eq_ as eq, assert_is_instance
+from numpy.testing import assert_array_equal
 
 
 from zarr.tests.test_attrs import TestAttributes
 from zarr.tests.test_core import TestArray
+from zarr.tests.test_hierarchy import TestGroup
 from zarr.sync import ThreadSynchronizer, ProcessSynchronizer
 from zarr.core import Array
 from zarr.attrs import Attributes
-from zarr.storage import init_array, TempStore
+from zarr.storage import init_array, TempStore, init_group, atexit_rmtree
 from zarr.compat import PY2
 from zarr.codecs import Zlib
+from zarr.hierarchy import Group
 
 
-class TestThreadSynchronizedAttributes(TestAttributes):
+if PY2:
+
+    class TemporaryDirectory(object):
+        def __init__(self):
+            self.name = tempfile.mkdtemp()
+            atexit.register(atexit_rmtree, self.name)
+
+else:
+    from tempfile import TemporaryDirectory
+
+
+class TestAttributesWithThreadSynchronizer(TestAttributes):
 
     def init_attributes(self, store, read_only=False):
         key = 'attrs'
@@ -34,7 +47,7 @@ def init_attributes(self, store, read_only=False):
                           read_only=read_only)
 
 
-class TestProcessSynchronizedAttributes(TestAttributes):
+class TestAttributesProcessSynchronizer(TestAttributes):
 
     def init_attributes(self, store, read_only=False):
         key = 'attrs'
@@ -46,7 +59,7 @@ def init_attributes(self, store, read_only=False):
                           read_only=read_only)
 
 
-def _append_data(arg):
+def _append(arg):
     z, i = arg
     import numpy as np
     x = np.empty(1000, dtype='i4')
@@ -55,22 +68,46 @@ def _append_data(arg):
     return z.shape
 
 
+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
+
+
 class MixinArraySyncTests(object):
 
+    def test_parallel_setitem(self):
+        n = 99
+
+        # setup
+        arr = self.create_array(shape=n * 1000, chunks=999, dtype='i4')
+        arr[:] = 0
+        pool = self.create_pool()
+
+        # parallel setitem
+        results = pool.map_async(_set_arange, zip([arr] * n, range(n)))
+        print(results.get())
+
+        assert_array_equal(np.arange(n * 1000), arr[:])
+
     def test_parallel_append(self):
+        n = 99
 
         # setup
-        arr = self.create_array(shape=1000, chunks=100, dtype='i4')
+        arr = self.create_array(shape=1000, chunks=999, dtype='i4')
         arr[:] = 0
-        pool = self.create_pool(cpu_count())
+        pool = self.create_pool()
 
-        results = pool.map_async(_append_data, zip([arr] * 39, range(1, 40, 1)))
+        # parallel append
+        results = pool.map_async(_append, zip([arr] * n, range(n)))
         print(results.get())
 
-        eq((40000,), arr.shape)
+        eq(((n+1)*1000,), arr.shape)
 
 
-class TestThreadSynchronizedArray(TestArray, MixinArraySyncTests):
+class TestArrayWithThreadSynchronizer(TestArray, MixinArraySyncTests):
 
     def create_array(self, read_only=False, **kwargs):
         store = dict()
@@ -93,17 +130,17 @@ def test_repr(self):
             for l1, l2 in zip(expect.split('\n'), actual.split('\n')):
                 eq(l1, l2)
 
-    def create_pool(self, size):
-        pool = ThreadPool(size)
+    def create_pool(self):
+        pool = ThreadPool(cpu_count())
         return pool
 
 
-class TestProcessSynchronizedArray(TestArray, MixinArraySyncTests):
+class TestArrayWithProcessSynchronizer(TestArray, MixinArraySyncTests):
 
     def create_array(self, read_only=False, **kwargs):
         store = TempStore()
         init_array(store, **kwargs)
-        synchronizer = ProcessSynchronizer(tempfile.TemporaryDirectory().name)
+        synchronizer = ProcessSynchronizer(TemporaryDirectory().name)
         return Array(store, synchronizer=synchronizer,
                      read_only=read_only, cache_metadata=False)
 
@@ -122,6 +159,113 @@ def test_repr(self):
             for l1, l2 in zip(expect.split('\n'), actual.split('\n')):
                 eq(l1, l2)
 
-    def create_pool(self, size):
-        pool = ProcessPool(size)
+    def create_pool(self):
+        pool = ProcessPool(cpu_count())
         return pool
+
+
+def _create_group(arg):
+    g, name = arg
+    h = g.create_group(name)
+    return h.name
+
+
+def _require_group(arg):
+    g, name = arg
+    h = g.require_group(name)
+    return h.name
+
+
+class MixinGroupSyncTests(object):
+
+    def test_parallel_create_group(self):
+
+        # setup
+        g = self.create_group()
+        pool = self.create_pool()
+
+        # parallel create group
+        n = 1000
+        results = pool.map_async(
+            _create_group, zip([g] * n, [str(i) for i in range(n)]))
+        print(results.get())
+
+        eq(n, len(g))
+
+    def test_parallel_require_group(self):
+
+        # setup
+        g = self.create_group()
+        pool = self.create_pool()
+
+        # parallel require group
+        n = 1000
+        results = pool.map_async(
+            _require_group, zip([g] * n, [str(i//10) for i in range(n)]))
+        print(results.get())
+
+        eq(n//10, len(g))
+
+
+class TestGroupWithThreadSynchronizer(TestGroup, MixinGroupSyncTests):
+
+    def create_group(self, store=None, path=None, read_only=False,
+                     chunk_store=None, synchronizer=None):
+        if store is None:
+            store, chunk_store = self.create_store()
+        init_group(store, path=path, chunk_store=chunk_store)
+        synchronizer = ThreadSynchronizer()
+        g = Group(store, path=path, read_only=read_only,
+                  chunk_store=chunk_store, synchronizer=synchronizer)
+        return g
+
+    def create_pool(self):
+        pool = ThreadPool(cpu_count())
+        return pool
+
+    def test_group_repr(self):
+        if not PY2:
+            g = self.create_group()
+            expect = 'Group(/, 0)\n' \
+                     '  store: dict; synchronizer: ThreadSynchronizer'
+            actual = repr(g)
+            for l1, l2 in zip(expect.split('\n'), actual.split('\n')):
+                eq(l1, l2)
+
+    def test_synchronizer_property(self):
+        g = self.create_group()
+        assert_is_instance(g.synchronizer, ThreadSynchronizer)
+
+
+class TestGroupWithProcessSynchronizer(TestGroup, MixinGroupSyncTests):
+
+    def create_store(self):
+        return TempStore(), None
+
+    def create_group(self, store=None, path=None, read_only=False,
+                     chunk_store=None, synchronizer=None):
+        if store is None:
+            store = TempStore()
+            chunk_store = None
+        init_group(store, path=path, chunk_store=chunk_store)
+        synchronizer = ProcessSynchronizer(TemporaryDirectory().name)
+        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())
+        return pool
+
+    def test_group_repr(self):
+        if not PY2:
+            g = self.create_group()
+            expect = 'Group(/, 0)\n' \
+                     '  store: TempStore; synchronizer: ProcessSynchronizer'
+            actual = repr(g)
+            for l1, l2 in zip(expect.split('\n'), actual.split('\n')):
+                eq(l1, l2)
+
+    def test_synchronizer_property(self):
+        g = self.create_group()
+        assert_is_instance(g.synchronizer, ProcessSynchronizer)