diff --git a/src/blosc2/embed_store.py b/src/blosc2/embed_store.py index 7ac77c54b..d3c69180e 100644 --- a/src/blosc2/embed_store.py +++ b/src/blosc2/embed_store.py @@ -288,17 +288,30 @@ def __setitem__( def __getitem__(self, key: str) -> blosc2.NDArray | SChunk | blosc2.ObjectArray | blosc2.BatchArray: """Retrieve a node from the embed store.""" - self._sync_metadata() - if key not in self._embed_map: - raise KeyError(f"Key '{key}' not found in the embed store.") - node_info = self._embed_map[key] - urlbase = node_info.get("urlbase", None) + if self._shared: + with self._backing_schunk.holding_lock(): + self._sync_metadata() + if key not in self._embed_map: + raise KeyError(f"Key '{key}' not found in the embed store.") + node_info = self._embed_map[key] + urlbase = node_info.get("urlbase", None) + if not urlbase: + offset = node_info["offset"] + length = node_info["length"] + serialized_data = bytes(self._store[offset : offset + length]) + else: + self._sync_metadata() + if key not in self._embed_map: + raise KeyError(f"Key '{key}' not found in the embed store.") + node_info = self._embed_map[key] + urlbase = node_info.get("urlbase", None) + if not urlbase: + offset = node_info["offset"] + length = node_info["length"] + serialized_data = bytes(self._store[offset : offset + length]) if urlbase: - urlpath = blosc2.URLPath(node_info["path"], urlbase=urlbase) - return blosc2.open(urlpath, mode="r") - offset = node_info["offset"] - length = node_info["length"] - serialized_data = bytes(self._store[offset : offset + length]) + # Outside the lock: opening a C2Array involves an HTTP round trip + return blosc2.open(blosc2.URLPath(node_info["path"], urlbase=urlbase), mode="r") # It is safer to copy data here, as the reference to the SChunk may disappear # Use from_cframe so we can deserialize either an NDArray or an SChunk return blosc2.from_cframe(serialized_data, copy=True) diff --git a/tests/test_locking.py b/tests/test_locking.py index 608faf925..106b9642e 100644 --- a/tests/test_locking.py +++ b/tests/test_locking.py @@ -1176,19 +1176,10 @@ def test_embed_store_cross_process_writers(tmp_path): keys = list(estore) assert "/seed" in keys for key in keys[-3:]: - try: - node = estore.get(key) # a concurrent delete cannot happen here - if node is None: - continue - data = node[:] - except RuntimeError: - # Same "listed before it is readable" window the None check - # covers, deeper in: the index already carries the key's - # (offset, length) but the backing schunk has not grown to - # cover it, so EmbedStore.__getitem__ raises "Error while - # getting the slice". Tolerated only while the writers are - # running -- every key is verified strictly once they exit. + node = estore.get(key) # a concurrent delete cannot happen here + if node is None: continue + data = node[:] assert len(data) == 10 nreads += 1 finally: @@ -1277,15 +1268,10 @@ def test_dict_store_cross_process_writers(tmp_path): for key in keys: if not key.endswith("ext3"): continue - try: - node = dstore.get(key) - if node is None: - continue - data = node[:] - except RuntimeError: - # See test_embed_store_cross_process_writers: the index - # carries the key before its bytes are readable. + node = dstore.get(key) + if node is None: continue + data = node[:] assert len(data) == 100 nreads += 1 finally: