Syncrhonization issue when creating groups in parallel

[jgkatz]

Minimal, reproducible code sample, a copy-pastable example if possible

import multiprocessing
import shutil
import zarr

def _par_create_group(arg):
    sync = zarr.ProcessSynchronizer('test.sync')
    zarr.open('test.zarr', mode='a', synchronizer=sync, cache_attrs=False)

for _ in range(100):
    pool = multiprocessing.Pool(multiprocessing.cpu_count())
    pool.map(_par_create_group, [0]*100)
    shutil.rmtree('test.zarr')

Problem description

When attempting to open a group in parallel for the first time in parallel, there appears to be a synchronization issue in the init_group function. It first checks for contains_group, and if False, initializes the group metadata. The process synchronizers are not used in this section, so it is possible to get into a race condition where the initial check comes up False but another process writes the metadata between the intial check and initializing metadata. The above code should produce:

multiprocessing.pool.RemoteTraceback: 
"""
Traceback (most recent call last):
  File "/usr/lib/python3.7/multiprocessing/pool.py", line 121, in worker
    result = (True, func(*args, **kwds))
  File "/usr/lib/python3.7/multiprocessing/pool.py", line 44, in mapstar
    return list(map(*args))
  File "test_zarr_sync.py", line 7, in _par_create_group
    zarr.open('test.zarr', mode='a', synchronizer=sync, cache_attrs=False)
  File "<venv_path>/lib/python3.7/site-packages/zarr/convenience.py", line 89, in open
    return open_group(store, mode=mode, **kwargs)
  File "<venv_path>/lib/python3.7/site-packages/zarr/hierarchy.py", line 1171, in open_group
    init_group(store, path=path, chunk_store=chunk_store)
  File "<venv_path>/lib/python3.7/site-packages/zarr/storage.py", line 473, in init_group
    chunk_store=chunk_store)
  File "<venv_path>/lib/python3.7/site-packages/zarr/storage.py", line 492, in _init_group_metadata
    raise ContainsGroupError(path)
zarr.errors.ContainsGroupError: path '' contains a group
"""

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "test_zarr_sync.py", line 11, in <module>
    pool.map(_par_create_group, [0]*100)
  File "/usr/lib/python3.7/multiprocessing/pool.py", line 268, in map
    return self._map_async(func, iterable, mapstar, chunksize).get()
  File "/usr/lib/python3.7/multiprocessing/pool.py", line 657, in get
    raise self._value
zarr.errors.ContainsGroupError: path "path '' contains a group" contains a group

Version and installation information

Please provide the following:

• Value of zarr.__version__: 2.5.0
• Value of numcodecs.__version__: 0.7.2
• Version of Python interpreter: 3.7
• Operating system (Linux/Windows/Mac): LInux
• How Zarr was installed (e.g., "using pip into virtual environment", or "using conda"): pip