Answer for the array a handle has itself written to

Review found three ways the write path could serve wrong data, and all
three are fixed here with a test apiece.

A handle reads api/info once when it opens the array, and `stamp` is
built from exactly the fields a write moves -- so a handle used to fill
an array kept its pre-fill stamp for good, and a Proxy given that handle
adopted a cache built while the chunks were still empty.  A write marks
the metadata stale now, and whatever asks reads it again.

`serves_blocks` weighs whether splitting a chunk into blocks would pay,
which is nothing to do with whether the frame has an index worth
reading -- but the two shared one memoized source, so touching the block
path first (max_ranges is a public property, and Proxy reads it) cached
None and left written_chunks raising NotRanged ever after.  The
judgement belongs at the call, not in the slot.

A complete array stamps `n<nonce>:<cbytes>` and a filling one whose
subscriber reports no mtime stamped the same string, which is the one
collision the whole design exists to prevent.  Both carry a marker now.

Also from the review: invalidating the index no longer empties what
index_state hands a cache, nor drops layouts a written chunk can never
invalidate; _forget_index takes the lock a source is built under; the
run-length tag is read in the host's byte order rather than always
little-endian; and the fill bench built a C2Array per chunk rather than
per writer, which charged every timed write an untallied api/info.  The
numbers move little -- 7.6x rather than 7.0x for eight writers over a
simulated WAN -- and the plan is corrected.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

Author: FrancescAlted

bench/ndarray/cat2-block-granularity.py

@@ -215,7 +215,7 @@ def _dataset(self):
             return target
         return self.server.subscriber
 
-    def do_POST(self):  # noqa: N802 (BaseHTTPRequestHandler's own spelling)
+    def do_POST(self):  # BaseHTTPRequestHandler's own spelling
         sub = self._dataset()
         endpoint = self.path.split("/")[2].split("?")[0]
         if endpoint != "chunk" or not sub.writable:
@@ -226,7 +226,7 @@ def do_POST(self):  # noqa: N802 (BaseHTTPRequestHandler's own spelling)
         status, answer = sub.write_chunk(nchunk, body)
         self._send(status, json.dumps(answer).encode())
 
-    def do_GET(self):  # noqa: N802 (BaseHTTPRequestHandler's own spelling)
+    def do_GET(self):  # BaseHTTPRequestHandler's own spelling
         sub = self._dataset()
         endpoint = self.path.split("/")[2]
         if endpoint == "info":
@@ -292,11 +292,16 @@ def _fetch(self, sub):
         )
 
 
-def stand_in(urlpath, streamed=False, target=None):
-    """Serve *urlpath* as ``@public/<name>``, and return (server, urlbase, path)."""
+def stand_in(urlpath, streamed=False):
+    """Serve *urlpath* as ``@public/<name>``, and return (server, urlbase, path).
+
+    The array a fill writes into is installed later, by `make_presize`, which
+    lays out a fresh one per timed fill; until then there is only the one served
+    here.
+    """
     server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), Handler)
     server.subscriber = Subscriber(urlpath, streamed)
-    server.target = None if target is None else Subscriber(target, writable=True)
+    server.target = None
     threading.Thread(target=server.serve_forever, daemon=True).start()
     urlbase = f"http://127.0.0.1:{server.server_address[1]}/"
     return server, urlbase, f"@public/{pathlib.Path(urlpath).name}"
@@ -476,33 +481,33 @@ def timed_fill(open_array, chunks, writers, latency, bandwidth):
     """
     tally = {"requests": 0, "bytes": 0}
     tally_lock = threading.Lock()
+    writers = max(writers, 1)
+    # One array per writer, built before the clock starts: opening one is an
+    # `api/info` of its own, and a writer opens once however many chunks it goes
+    # on to send.  Building them inside the timing would charge every chunk for a
+    # round trip no writer actually makes
+    arrays = [open_array() for _ in range(writers)]
+    work = list(enumerate(chunks))
+    shares = [work[index::writers] for index in range(writers)]
 
-    def write(nchunk_and_chunk):
-        nchunk, chunk = nchunk_and_chunk
-        array = open_array()
-        original = array.update_chunk
-
-        def charged(n, payload):
+    def run(assignment):
+        array, share = assignment
+        for nchunk, chunk in share:
             if latency:
                 time.sleep(latency)
             if bandwidth:
-                time.sleep(len(payload) / bandwidth)
-            answer = original(n, payload)
+                time.sleep(len(chunk) / bandwidth)
+            array.update_chunk(nchunk, chunk)
             with tally_lock:
                 tally["requests"] += 1
-                tally["bytes"] += len(payload)
-            return answer
-
-        return charged(nchunk, chunk)
+                tally["bytes"] += len(chunk)
 
-    work = list(enumerate(chunks))
     start = time.perf_counter()
-    if writers <= 1:
-        for item in work:
-            write(item)
+    if writers == 1:
+        run((arrays[0], work))
     else:
         with concurrent.futures.ThreadPoolExecutor(max_workers=writers) as pool:
-            list(pool.map(write, work))
+            list(pool.map(run, zip(arrays, shares, strict=True)))
     return time.perf_counter() - start, tally["requests"], tally["bytes"]
 
 

doc/reference/c2array.rst

@@ -47,6 +47,9 @@ own offsets, which is one range read and no endpoint of its own.
     .. automethod:: __getitem__
 
 
+.. autoclass:: ChunkAlreadyWritten
+
+
 .. _C2NDSource:
 
 C2NDSource class

plans/cat2-concurrent-writers.md

@@ -464,8 +464,8 @@ of 1.76 MB:
 
 | | | |
 |---|---|---|
-| fill, serial | 247.7 ms/chunk | one round trip apiece |
-| fill, 8 writers at once | **35.5 ms/chunk** | **7.0x** |
+| fill, serial | 244.0 ms/chunk | one round trip apiece |
+| fill, 8 writers at once | **32.2 ms/chunk** | **7.6x** |
 | store into an empty slot | 0.91 ms | appended; no other chunk moves |
 | store over a live chunk | 2.96 ms | 3.3x, rewriting the 5.29 MB after it |
 | `written_chunks()` over HTTP | 2.47 ms | one range read of the offsets |
@@ -475,7 +475,7 @@ of 1.76 MB:
 The concurrency figure is the one worth having: the subscriber serializes the
 writes themselves, since each takes the frame's exclusive lock, so what overlaps
 is the round trip — which over a WAN is nearly all of it, and over loopback is
-nearly none (1.2x there).  The rewrite ratio is this dataset's and grows with
+none at all (1.0x there: 1.9 ms a chunk either way).  The rewrite ratio is this dataset's and grows with
 whatever payload follows the chunk; the same measurement on a 110 MB frame ran
 21.2 ms against 0.5 ms.
 

src/blosc2/c2array.py

@@ -565,6 +565,9 @@ def __init__(self, path: str, /, urlbase: str | None = None, auth_token: str | N
         # dataset cannot be read in ranges) or a C2NDSource
         self._block_source = _UNTRIED
         self._block_lock = threading.Lock()
+        # Set when this handle writes: `meta` describes the array as it was read,
+        # and a write of its own moves what `stamp` and `vlmeta` are built from
+        self._meta_stale = False
         # An index a `Proxy` handed over before the source existed; see adopt_index
         self._pending_index = None
 
@@ -854,9 +857,25 @@ def written_chunks(self) -> np.ndarray:
 
     def _forget_index(self) -> None:
         """Drop what was read of a frame that has since been written to."""
-        source = self._block_source
-        if source is not _UNTRIED and source is not None:
-            source.invalidate_index()
+        # The metadata as well as the index: `meta` is read once when the array
+        # is opened, so a handle that goes on to write would otherwise answer for
+        # the array as it was before its own writes -- and `stamp` is built from
+        # exactly the fields a write moves.  Read again when something asks,
+        # rather than here, so a writer that never asks pays no request
+        self._meta_stale = True
+        with self._block_lock:
+            # Under the lock a source being built right now is invalidated after
+            # it is built, rather than missed entirely for holding a header this
+            # write has already moved
+            source = self._block_source
+            if source is not _UNTRIED and source is not None:
+                source.invalidate_index()
+
+    def _refresh_meta(self) -> None:
+        """Read `api/info` again, if this handle has written since it last did."""
+        if self._meta_stale:
+            self.meta = info(self.path, self.urlbase, auth_token=self.auth_token)
+            self._meta_stale = False
 
     # -- Block-granular reads.  A :ref:`Proxy` uses these to fetch the blocks a
     # slice touches instead of whole chunks, wherever that is the cheaper way
@@ -897,16 +916,21 @@ def stamp(self) -> str | None:
         which leaves the cache checked on its geometry alone, as every source
         without a stamp is.
         """
+        self._refresh_meta()
         vlmeta = self.meta.get("schunk", {}).get("vlmeta") or {}
         nonce = vlmeta.get("fill_nonce")
         cbytes = self.meta.get("schunk", {}).get("cbytes", "")
-        if nonce is not None and vlmeta.get("fill_state", "filling") != "filling":
-            # Complete: nothing can write to it again, so nothing here need move
-            return f"n{nonce}:{cbytes}"
         mtime = self.meta.get("mtime")
-        if mtime is None:
-            return None if nonce is None else f"n{nonce}:{cbytes}"
-        return f"{mtime}:{cbytes}" if nonce is None else f"n{nonce}:{mtime}:{cbytes}"
+        if nonce is None:
+            return None if mtime is None else f"{mtime}:{cbytes}"
+        # `c` and `f` keep the two apart whatever the rest holds: a complete array
+        # and a filling one must never stamp the same, or a cache of the second
+        # is adopted against the first and serves the zeros it holds for the
+        # chunks nobody had written yet
+        if vlmeta.get("fill_state", "filling") != "filling":
+            # Complete: nothing can write to it again, so nothing here need move
+            return f"n{nonce}:c:{cbytes}"
+        return f"n{nonce}:f:{cbytes}" if mtime is None else f"n{nonce}:f:{mtime}:{cbytes}"
 
     @property
     def blocks_per_chunk(self) -> int:
@@ -964,8 +988,14 @@ def block_source(self) -> C2NDSource | None:
         be permanent: a subscriber that streams this dataset answers a range
         request with the whole body, so retrying would pay a full download to
         rediscover the same answer.
+
+        A frame whose chunks are too small to be worth taking apart says no here
+        without building anything, and without remembering that it said so: the
+        judgement is about *blocks*, and the same frame's index is still worth
+        reading.  Deciding it at the call rather than caching it is what keeps
+        the two questions from answering each other.
         """
-        return self._source(require_blocks=True)
+        return self._source() if self.serves_blocks else None
 
     def _index_source(self) -> C2NDSource | None:
         """The same reader, built for any stored frame however small its chunks.
@@ -975,29 +1005,29 @@ def _index_source(self) -> C2NDSource | None:
         and they still say which chunks hold anything.  Whatever is built here is
         the source the block path uses too -- there is only ever one.
         """
-        return self._source(require_blocks=False)
+        return self._source() if self._reports_geometry else None
 
-    def _source(self, require_blocks: bool) -> C2NDSource | None:
+    def _source(self) -> C2NDSource | None:
+        """The one source, built once, whichever question asked for it first."""
         if self._block_source is _UNTRIED:
             with self._block_lock:
                 if self._block_source is _UNTRIED:
-                    self._block_source = self._open_block_source(require_blocks)
+                    self._block_source = self._open_block_source()
         # A failure that says nothing about the dataset leaves it _UNTRIED, so the
         # next fetch asks again; this one keeps to whole chunks either way
         return None if self._block_source is _UNTRIED else self._block_source
 
-    def _open_block_source(self, require_blocks: bool = True):
+    def _open_block_source(self):
         """Decide, at whatever cost it takes, whether this dataset serves ranges.
 
         None for a dataset that does not serve ranges, which is an answer for
         good; `_UNTRIED` for a subscriber that could not say, which is not.
         """
         httpx = _httpx()
-        # What `api/info` alone rules out -- a dataset the subscriber computes, a
-        # frame of chunks too small to take apart -- costs no request to find out.
-        # The second of those only bars the block path: the index is worth a read
-        # whatever the chunks cost, which is what `require_blocks` says
-        if not (self.serves_blocks if require_blocks else self._reports_geometry):
+        # A dataset the subscriber computes has no frame to read at all, and
+        # `api/info` says so for free.  Whether its chunks are worth taking apart
+        # is a separate judgement, made by whoever asks -- see `block_source`
+        if not self._reports_geometry:
             return None
         # Whether a dataset that reports a geometry is *served* from a file is
         # something only the answer to a range request can say: an HDF5 leaf or a
@@ -1179,7 +1209,13 @@ def cratio(self) -> float:
 
     @property
     def vlmeta(self) -> dict:
-        """The variable-length metadata f the remote array"""
+        """The variable-length metadata of the remote array.
+
+        Read again where this handle has written since it last looked: a fill
+        records itself here, so a writer asking what it just did would otherwise
+        be told what was true before it started.
+        """
+        self._refresh_meta()
         return self.meta["schunk"]["vlmeta"]
 
     @property

src/blosc2/proxy_source.py

@@ -684,10 +684,12 @@ def _frame_index(self) -> tuple[np.ndarray, np.ndarray]:
         with self._index_lock:
             if self._stale:
                 # A write moved the frame's length and its payload extent, and the
-                # offsets are found through both, so the header is read first
+                # offsets are found through both, so the header is read first, and
+                # the offsets it locates are read again after it
                 raw, self._header, self._head = _read_frame_header(self.read_range)
                 self._header_len = len(raw)
                 self._chunksize = self._header[8]
+                self._index = None
                 self._stale = False
             if self._index is None:
                 offsets = _read_frame_offsets(self.read_range, self._header, self._head, self._header_len)
@@ -722,8 +724,11 @@ def written_chunks(self) -> np.ndarray:
         """
         offsets = self._offsets
         # A view, not a cast: the tag lives in the top byte of an offset whose
-        # sign bit is what marks it as run-length in the first place
-        kinds = (offsets.view("<u8") >> 56) & 0x7
+        # sign bit is what marks it as run-length in the first place.  Viewed as
+        # the native unsigned type, not as a little-endian one -- the offsets are
+        # read in the host's order, so naming a byte order here would read the
+        # tag out of the wrong end of each word on a big-endian machine
+        kinds = (offsets.view(np.uint64) >> np.uint64(56)) & np.uint64(0x7)
         return ~((offsets < 0) & (kinds == _SPECIAL_UNINIT))
 
     def invalidate_index(self) -> None:
@@ -744,8 +749,14 @@ def invalidate_index(self) -> None:
         writing.  A frame that nobody mutates never needs this.
         """
         with self._index_lock:
-            self._index = None
-            self._layouts.clear()
+            # What was read stays until something reads again: `index_state` hands
+            # it to a cache that a stamp already guards, and emptying it here
+            # would overwrite a good index with nothing at all.
+            #
+            # The layouts stay for good.  A chunk gets one only once it has been
+            # read, which under the write-once contract this exists for means it
+            # holds content and can never be written again; a chunk that was
+            # empty when the index was read has no layout to be wrong about.
             self._stale = True
 
     @property
@@ -927,10 +938,12 @@ def _special_chunk(self, offset: int) -> bytes:
         nitems = self._chunksize // self._dtype.itemsize
         if kind == _SPECIAL_NAN:
             data = np.full(nitems, np.nan, dtype=self._dtype)
-        else:
+        elif kind in (_SPECIAL_ZERO, _SPECIAL_UNINIT):
             # A run of zeros; an uninitialized chunk has no defined content, and
             # zeros is what reading one locally hands back too
             data = np.zeros(nitems, dtype=self._dtype)
+        else:
+            raise NotImplementedError(f"chunk offset {offset} codes run-length value {kind}")
         # The blocksize has to be the container's: left to choose, blosc2 takes
         # the whole chunk, and the cache then rejects the chunk we hand it
         return blosc2.compress2(

tests/ndarray/test_c2array_writes.py

@@ -474,3 +474,60 @@ def test_a_cache_of_a_complete_array_survives_a_second_run(subscriber, tmp_path)
     proxy = blosc2.Proxy(blosc2.C2Array("run.b2nd", urlbase=array.urlbase), urlpath=cache, mode="a")
     np.testing.assert_array_equal(proxy[:], expected)
     assert not [entry for entry in sub.log if entry[0] in ("chunk", "fetch")]
+
+
+def test_a_handle_that_writes_stamps_what_it_wrote(subscriber):
+    """A writer's own view of the array has to move when the array does.
+
+    `meta` is read when the array is opened, and `stamp` is built from exactly
+    the fields a write moves, so a handle that goes on to write would otherwise
+    answer for the array as it was before its own writes -- and a `Proxy` given
+    that handle would adopt a cache built against them.
+    """
+    array, sub = subscriber
+    before = array.stamp
+    array.update_chunk(0, _chunk(0))
+    assert array.stamp != before
+    assert array.stamp == blosc2.C2Array("run.b2nd", urlbase=array.urlbase).stamp
+
+
+def test_asking_about_blocks_does_not_close_the_door_on_the_index(subscriber):
+    """Two questions, one source, and the answer to one must not answer the other.
+
+    `serves_blocks` weighs whether splitting a chunk into blocks would pay, which
+    a frame of small chunks fails; reading the frame's index is worth doing
+    anyway.  Deciding that at the call rather than remembering it is what keeps
+    the block path from shutting the index path down.
+    """
+    array, sub = subscriber
+    assert not array.serves_blocks  # chunks here are far under BLOCK_MIN_CBYTES
+    assert array.max_ranges == 1  # the block path, asked first, and declining
+    assert array.block_source() is None
+    assert list(array.written_chunks()) == [False] * NCHUNKS  # still answerable
+
+
+def test_a_filling_stamp_can_never_read_as_a_complete_one(subscriber):
+    """The two branches must not be able to produce the same string.
+
+    A cache built while chunks were unwritten holds the zeros they read as; if
+    the completed array stamped the same, that cache would be adopted against it
+    and those zeros served as data.
+    """
+    array, sub = subscriber
+    array.update_chunk(0, _chunk(0))
+    filling = blosc2.C2Array("run.b2nd", urlbase=array.urlbase).stamp
+    assert ":f:" in filling
+    for nchunk in range(1, NCHUNKS):
+        array.update_chunk(nchunk, _chunk(nchunk))
+    complete = blosc2.C2Array("run.b2nd", urlbase=array.urlbase).stamp
+    assert ":c:" in complete
+    assert complete != filling
+
+    # ... including when the subscriber reports no mtime at all, which is what
+    # left the two able to collide
+    handle = blosc2.C2Array("run.b2nd", urlbase=array.urlbase)
+    handle.meta["mtime"] = None
+    handle.meta["schunk"]["vlmeta"] = {"fill_nonce": "abc", "fill_state": "filling"}
+    unfinished = handle.stamp
+    handle.meta["schunk"]["vlmeta"] = {"fill_nonce": "abc", "fill_state": "complete"}
+    assert handle.stamp != unfinished