diff --git a/RELEASE_NOTES.md b/RELEASE_NOTES.md index d17e4a92e..9674c4320 100644 --- a/RELEASE_NOTES.md +++ b/RELEASE_NOTES.md @@ -8,7 +8,8 @@ XXX version-specific blurb XXX * New `blosc2[fsspec]` extra: `blosc2.open()`, `save_array()` and `save_tensor()` accept any [fsspec](https://filesystem-spec.readthedocs.io) URL — `s3://`, - `gs://`, `zip://`, chained ones like `zip://inner.b2nd::s3://bucket/a.zip`. + `gs://`, `https://`, `zip://`, chained ones like + `zip://inner.b2nd::s3://bucket/a.zip`. `open()` reads the container whole, or through a staleness-checked local copy with `cache_storage=` (which is what covers `.b2d` stores, sparse frames, `offset` and `mmap_mode`), or a piece at a time with `lazy=True`, which @@ -22,6 +23,33 @@ XXX version-specific blurb XXX which an object store has no way to serve), so constructors given a URL now say that instead of failing deep in C. +* A `C2Array` can be written to a chunk at a time, which is how several + processes fill one remote array at once: `update_chunk()` (and its async + `aupdate_chunk()`) posts one compressed chunk into a slot of a pre-sized + array, and `written_chunks()` says which slots hold anything yet. The array is + laid out with `blosc2.uninit()` and uploaded -- a couple of hundred bytes + whatever its size -- and each slot is written once: a second write raises + `blosc2.ChunkAlreadyWritten`, which is the whole of the coordination between + writers. Writing into an empty slot appends to the frame and moves no other + chunk, so a fill is cheap and a concurrent reader's cached offsets stay good. + Needs a Caterva2 subscriber that serves the endpoint. + +* `C2Array.stamp`, which is what a `Proxy` checks its cache against, now names + *which* array it is as well as whether it has changed. A subscriber writes a + nonce into a filled array's vlmeta, so a cache is no longer served against a + different array that came to sit at the same path with the same size and + mtime; and a complete array — every chunk written, so every further write + refused — is stamped without its mtime, so a cache of it survives a republish + or a copy instead of being thrown away. Arrays that were never filled a chunk + at a time are stamped exactly as before. + + A `Proxy` now calls `C2Array.refresh_stamp()` before judging its cache, which + reads `api/info` once for an array that could still be written to. A handle + reads that once when it is opened and, of itself, never again, so one that has + outlived someone else's chunks would otherwise hand over the stamp of the + array as it was — which its cache matches and the remote bytes no longer do. A + complete array costs nothing here: nothing can write to one. + * `Proxy.fetch()` takes a `max_concurrency=` argument, and reads it from the source when the source has one, so `blosc2.open(url, lazy=True, max_concurrency=...)` overlaps its chunk fetches in a thread pool. Ordinary diff --git a/bench/ndarray/cat2-block-granularity.py b/bench/ndarray/cat2-block-granularity.py index 0d2e3ed94..2983d3211 100644 --- a/bench/ndarray/cat2-block-granularity.py +++ b/bench/ndarray/cat2-block-granularity.py @@ -39,6 +39,9 @@ python cat2-block-granularity.py @public/examples/kevlar-tomo.b2nd \\ --urlbase https://cat2.cloud/demo + # ... and what filling a pre-sized array costs, one chunk per request + python cat2-block-granularity.py mydata.b2nd --write + # ... an authenticated dataset python cat2-block-granularity.py @personal/mine.b2nd --urlbase http://localhost:8000 \\ --username me@example.com --password foobar11 @@ -67,17 +70,44 @@ object store and about wrong for one subscriber, and is why ``multipart`` can come out behind ``blocks`` there while it wins against the real thing. +``--write`` measures the other direction: an array is laid out empty and filled +a chunk at a time, which is how several writers fill one array at once. Three +things, and the first is the only one that goes over the wire: + +- the **fill**, serial and then ``--concurrency`` writers at once. The + subscriber serializes the writes themselves -- each takes the frame's + exclusive lock -- so what overlaps is the round trip, and the gain is whatever + share of a write that was. Over loopback it is almost none; put a network in + front with ``--latency-ms`` and it is most of it; +- what the **server pays to store one chunk**, into an empty slot and over a live + one, timed locally where a round trip would bury the difference. A slot + holding nothing is appended past the offsets and moves no other chunk; one + holding a chunk has every byte of payload after it read and written back. + That difference is why a fill writes each slot once and refuses a second write; +- what **reading the progress** costs, from the frame's offsets against walking + its chunks. The offsets are one decompress whatever the count; the walk is a + read per chunk, so the two cross over as an array grows. + +Against a real subscriber ``--write`` needs ``--write-target``: an empty +pre-sized array to fill, since laying one out is not this script's business on +someone else's server. Only the serial fill runs there -- a slot is written +once, so a second timed fill needs a second array. + Bytes counted are payload: the multipart envelope (about a hundred bytes per part) and the HTTP headers of every request are not in them. """ import argparse +import concurrent.futures import http.server +import itertools import json import math import pathlib +import shutil import statistics import struct +import tempfile import threading import time @@ -92,17 +122,57 @@ # +UNINIT = 0x4 +"""What a frame codes in a chunk's flags byte for a slot never written to.""" + + class Subscriber: - """Caterva2's three read endpoints over one local .b2nd file.""" + """Caterva2's read endpoints over one local .b2nd file, and its write one.""" - def __init__(self, urlpath, streamed=False): + def __init__(self, urlpath, streamed=False, writable=False): self.path = pathlib.Path(urlpath) + self.name = self.path.name self.size = self.path.stat().st_size - self.array = blosc2.open(str(self.path)) + # A writable dataset is opened once, for the life of the server, and + # written through that one handle: a second handle over a frame this one + # writes leaves it unreadable, and says nothing while doing so + self.writable = writable + self.array = blosc2.open(str(self.path), mode="a" if writable else "r", locking=writable) + self.lock = threading.Lock() # A dataset the subscriber would compute rather than store: served by a # body builder, which has no way to honour a Range self.streamed = streamed + def close(self): + """Let go of the file this held open, so the scratch tree can be removed. + + A writable subscriber keeps one handle for its whole life, and a run that + fills several arrays leaves one behind per array otherwise -- still + holding files that `shutil.rmtree` then unlinks under them. + """ + self.array = None + + def write_chunk(self, nchunk, chunk): + """Caterva2's write contract: one chunk, into a slot that holds none. + + The refusal is the whole of the coordination between writers, and the + check is O(1) -- a lazy chunk is its header, where walking the array + would make a fill cost the square of its length. + """ + with self.lock: + schunk = self.array.schunk + if not 0 <= nchunk < schunk.nchunks: + return 404, {"detail": "no such chunk"} + nbytes, _, blocksize = blosc2.get_cbuffer_sizes(chunk) + if nbytes != schunk.chunksize or blocksize != schunk.blocksize: + return 400, {"detail": "the chunk does not match the array's geometry"} + with schunk.holding_lock(): + if (schunk.get_lazychunk(nchunk)[31] >> 4) & 0x7 != UNINIT: + return 409, {"detail": f"chunk {nchunk} was already written"} + schunk.update_chunk(nchunk, chunk) + self.size = self.path.stat().st_size + return 200, {"nchunk": nchunk} + def meta(self): schunk = self.array.schunk return { @@ -143,8 +213,30 @@ def _send(self, status, body, headers=()): self.end_headers() self.wfile.write(body) - def do_GET(self): # noqa: N802 (BaseHTTPRequestHandler's own spelling) - sub = self.server.subscriber + def _dataset(self): + """Which of the served datasets this request names. + + One of them until a fill is being measured, when there is a second: the + array being filled, which is not the array being read. + """ + target = getattr(self.server, "target", None) + if target is not None and self.path.split("?")[0].endswith(target.name): + return target + return self.server.subscriber + + 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: + self._send(404, b"") + return + nchunk = int(self.path.split("nchunk=")[1]) + body = self.rfile.read(int(self.headers.get("Content-Length", 0))) + status, answer = sub.write_chunk(nchunk, body) + self._send(status, json.dumps(answer).encode()) + + def do_GET(self): # BaseHTTPRequestHandler's own spelling + sub = self._dataset() endpoint = self.path.split("/")[2] if endpoint == "info": self._send(200, json.dumps(sub.meta()).encode()) @@ -210,9 +302,15 @@ def _fetch(self, sub): def stand_in(urlpath, streamed=False): - """Serve *urlpath* as ``@public/``, and return (server, urlbase, path).""" + """Serve *urlpath* as ``@public/``, 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 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}" @@ -370,6 +468,91 @@ def timed_slice(open_array, item, mode, concurrency, latency, bandwidth): blosc2.proxy_source.BLOCK_MIN_CBYTES = threshold +def fill_chunks(source, limit): + """The dataset's own compressed chunks, which is what a fill would carry. + + Real chunks rather than synthetic ones, so the bytes on the wire and the + work the server does storing them are the dataset's own. Capped, because a + fill is timed per chunk and a large array would only repeat the measurement. + """ + nchunks = math.prod(math.ceil(s / c) for s, c in zip(source.shape, source.chunks, strict=True)) + # `get_chunk` is the one both a local array and a `C2Array` answer, so the + # bytes are the dataset's own whether it is a file here or a dataset there + return [source.get_chunk(n) for n in range(min(nchunks, limit))] + + +def timed_fill(open_array, chunks, writers, latency, bandwidth): + """Write *chunks* into a pre-sized array, and say what it cost. + + One `C2Array` per writer, as separate processes would have. What overlaps + is the round trip: the subscriber serializes the writes themselves, since + each one takes the frame's exclusive lock. + """ + 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 run(assignment): + array, share = assignment + for nchunk, chunk in share: + if latency: + time.sleep(latency) + if bandwidth: + time.sleep(len(chunk) / bandwidth) + array.update_chunk(nchunk, chunk) + with tally_lock: + tally["requests"] += 1 + tally["bytes"] += len(chunk) + + start = time.perf_counter() + if writers == 1: + run((arrays[0], work)) + else: + with concurrent.futures.ThreadPoolExecutor(max_workers=writers) as pool: + list(pool.map(run, zip(arrays, shares, strict=True))) + return time.perf_counter() - start, tally["requests"], tally["bytes"] + + +def local_write_cost(presize, chunks, reps): + """What the *server* pays to store a chunk, into an empty slot and over a live one. + + Measured on local files rather than over HTTP: this is the difference the + write-once rule buys, and a round trip would bury it. A slot that holds + nothing is appended to and moves no other chunk; one that holds a chunk is + written in place, and every byte of payload after it is read and written back + to close the gap the old chunk left. + + The rewrite has to carry a chunk of a *different* compressed size, or it + measures the wrong thing: replacing a chunk with bytes of its own length + leaves nothing to close, and the frame skips the move entirely. None when + the dataset has no two chunks that differ in size to do it with. + """ + middle = len(chunks) // 2 + other = next((c for c in chunks if len(c) != len(chunks[middle])), None) + empty, live = [], [] + for _ in range(reps): + path = presize() + array = blosc2.open(path, mode="a", locking=True) + for nchunk, chunk in enumerate(chunks): + start = time.perf_counter() + array.schunk.update_chunk(nchunk, chunk) + empty.append(time.perf_counter() - start) + if other is not None: + # Every slot holds something now, so this one compacts instead + start = time.perf_counter() + array.schunk.update_chunk(middle, other) + live.append(time.perf_counter() - start) + del array + return statistics.median(empty), (statistics.median(live) if live else None) + + def connection_setup(urlbase, path, token, reps): """What a request costs before any bytes move, pooled against a client each. @@ -407,6 +590,18 @@ def main(): action="store_true", help="stand-in only: serve the dataset the way a computed one is served", ) + parser.add_argument( + "--write", + action="store_true", + help="also measure filling a pre-sized array a chunk at a time", + ) + parser.add_argument( + "--write-target", + help="with --urlbase and --write: an empty pre-sized array to fill (else one is laid out)", + ) + parser.add_argument( + "--fill-chunks", type=int, default=10, help="how many chunks a timed fill writes (default: 10)" + ) parser.add_argument("--concurrency", type=int, default=8, help="parallel requests (default: 8)") parser.add_argument("--reps", type=int, default=5, help="timed repetitions (default: 5)") parser.add_argument("--max-mb", type=float, default=200, help="skip patterns fetching more than this") @@ -421,20 +616,51 @@ def main(): server = None if args.urlbase: urlbase, path = args.urlbase, args.dataset + if args.write and not args.write_target: + parser.error("--write against a subscriber needs --write-target: an empty array to fill") else: server, urlbase, path = stand_in(args.dataset, args.streamed) token = args.token if args.username: token = c2array.login(args.username, args.password, urlbase) + scratch = tempfile.mkdtemp(prefix="cat2-fill-") if args.write and server else None + presize = make_presize(args.dataset, scratch, server) if scratch else None try: - report(args, urlbase, path, token) + report(args, urlbase, path, token, presize) finally: if server is not None: server.shutdown() + if scratch: + shutil.rmtree(scratch, ignore_errors=True) + + +def make_presize(source_path, scratch, server): + """Lay out an empty array of the dataset's geometry, ready to be filled. + + A fresh one per call: a slot is written once, so a second timed fill needs a + second array. Costs a couple of hundred bytes whatever the geometry -- an + unwritten chunk lives in the offsets and nowhere else. + """ + source = blosc2.open(str(source_path)) + counter = itertools.count() + def presize(serve=False): + path = str(pathlib.Path(scratch) / f"fill-{next(counter)}.b2nd") + laid_out = blosc2.uninit( + source.shape, dtype=source.dtype, chunks=source.chunks, blocks=source.blocks, urlpath=path + ) + del laid_out # the server's handle is to be the only one over this file + if serve: + if server.target is not None: + server.target.close() # its handle is done with; the next array gets its own + server.target = Subscriber(path, writable=True) + return path + + return presize -def report(args, urlbase, path, token): + +def report(args, urlbase, path, token, presize=None): latency, bandwidth = args.latency_ms / 1e3, args.bandwidth_mbs * 1e6 def open_array(): @@ -465,6 +691,8 @@ def open_array(): " A proxy over this fetches whole chunks, exactly as it always did." ) _time_patterns(args, open_array, array, ["chunks"], latency, bandwidth) + if args.write: + _fill_section(args, urlbase, token, path, presize, latency, bandwidth) return source.read_ranges([(0, 16), (64, 16)]) # two spans that cannot merge into one print( @@ -495,6 +723,101 @@ def open_array(): f"a client per request ({fresh / pooled:.1f}x)" ) _time_patterns(args, open_array, array, ["chunks", "blocks", "multipart"], latency, bandwidth, plans) + if args.write: + _fill_section(args, urlbase, token, path, presize, latency, bandwidth) + + +def _fill_section(args, urlbase, token, path, presize, latency, bandwidth): + """The write path, over the same connection the reads were measured on.""" + local = args.dataset if presize is not None else None + source = blosc2.open(str(local)) if local else c2array.C2Array(path, urlbase, token) + _report_fill(args, urlbase, token, source, presize, args.write_target, latency, bandwidth) + + +def _report_fill(args, urlbase, token, source, presize, target_path, latency, bandwidth): + """What filling a pre-sized array costs, and what the write-once rule buys.""" + chunks = fill_chunks(source, args.fill_chunks) + payload = sum(len(chunk) for chunk in chunks) + print( + f"\n fill: {len(chunks)} chunks, {payload / 1e6:.2f} MB of the dataset's own " + f"compressed bytes\n" + f" {'writers':16s} {'requests':>8s} {'bytes':>10s} {'total':>9s} {'per chunk':>11s}" + ) + runs = [("serial", 1)] + if args.concurrency > 1: + runs.append((f"{args.concurrency} at once", args.concurrency)) + serial = None + filled = filled_path = None + for label, writers in runs: + if presize is None and serial is not None: + # A real target's slots are one-shot, and the bench does not lay out + # a second array on someone else's server + print(f" {'(a second fill needs a second empty array)':16s}") + break + path = target_path + if presize is not None: + filled_path = presize(serve=True) + path = f"@public/{pathlib.Path(filled_path).name}" + + def open_array(remote=path): + return c2array.C2Array(remote, urlbase=urlbase, auth_token=token) + + elapsed, requests, nbytes = timed_fill(open_array, chunks, writers, latency, bandwidth) + # `is None`, not falsiness: a fill fast enough to measure as 0.0 is a + # measurement, and taking it for "not measured yet" would make the + # concurrent run its own baseline and report a speedup of 1.0x + serial = elapsed if serial is None else serial + filled = open_array() + speedup = f" {serial / elapsed:.1f}x" if writers > 1 else "" + print( + f" {label:16s} {requests:8d} {nbytes / 1e6:9.2f} MB {elapsed:8.3f} s " + f"{elapsed / len(chunks) * 1e3:9.1f} ms{speedup}" + ) + + if presize is not None: + empty, live = local_write_cost(lambda: presize(serve=False), chunks, args.reps) + # What the rewrite has to shift, which is what its cost is made of: the + # ratio below is this dataset's, and grows with whatever follows a chunk + tail = sum(len(chunk) for chunk in chunks[len(chunks) // 2 + 1 :]) + rewrite = ( + f" over a live chunk {live * 1e3:8.2f} ms {live / empty:.1f}x here, reading and " + f"writing back the {tail / 1e6:.2f} MB after it" + if live is not None + else " over a live chunk n/a every chunk here compresses to the same size, " + "which is the case that never moves" + ) + print( + f"\n what the server pays to store one chunk (local, median of {args.reps})\n" + f" into an empty slot {empty * 1e3:8.2f} ms appended past the offsets; " + f"no other chunk moves\n{rewrite}" + ) + + if filled is not None: + start = time.perf_counter() + written = filled.written_chunks() + remote = time.perf_counter() - start + print( + f"\n reading how far a fill has got ({int(written.sum())}/{written.size} written)\n" + f" written_chunks() {remote * 1e3:8.2f} ms over HTTP: the frame's header, " + "then the offsets it locates" + ) + if filled_path is not None: + # The same question the server asks itself on every write, both ways + # round and both local, since one of them is not a thing to ask remotely + blosc2.FsspecNDSource(filled_path).written_chunks() # fsspec's first use is its own cost + start = time.perf_counter() + offsets = blosc2.FsspecNDSource(filled_path).written_chunks() + index = time.perf_counter() - start + array = blosc2.open(filled_path) + start = time.perf_counter() + walked = sum(1 for info in array.schunk.iterchunks_info() if info.special.name != "UNINIT") + walk = time.perf_counter() - start + print( + f" ... the same, local {index * 1e3:8.2f} ms one decompress of the offsets, " + f"whatever the count\n" + f" iterchunks_info() {walk * 1e3:8.2f} ms {walk / offsets.size * 1e6:.1f} us per " + f"chunk ({walked} written), which is what grows with the array" + ) def _no_chunks(array): diff --git a/doc/guides/index.rst b/doc/guides/index.rst index 5022686cd..3206bfb0f 100644 --- a/doc/guides/index.rst +++ b/doc/guides/index.rst @@ -12,6 +12,7 @@ Topics :maxdepth: 1 optimization_tips + remote_arrays sharing_across_processes pandas_engine diff --git a/doc/guides/remote_arrays.md b/doc/guides/remote_arrays.md new file mode 100644 index 000000000..ffbb92053 --- /dev/null +++ b/doc/guides/remote_arrays.md @@ -0,0 +1,166 @@ +# Working with Remote Arrays + +A Blosc2 array that lives on a server does not have to be downloaded to be used. Blosc2 opens it where it is, fetches only the pieces a slice touches, and keeps those in a local cache so the next run starts from them. + +## Three ways in + +| Where the array lives | How to open it | +|---|---| +| Any URL fsspec reaches — `s3://`, `gs://`, `https://`, `zip://`… | `blosc2.open(url, lazy=True)` | +| A [Caterva2](https://ironarray.io/caterva2) subscriber | `blosc2.C2Array(path, urlbase=...)` | +| Anything else | A `read_range()` of your own — see [Your own transport](#your-own-transport) | + +```python +import blosc2 + +# An object store, a web server, a zip on either of them +a = blosc2.open("s3://bucket/big.b2nd", lazy=True) + +# A Caterva2 subscriber +b = blosc2.C2Array( + "@public/examples/lung-jpeg2000_10x.b2nd", urlbase="https://cat2.cloud/demo" +) + +a.shape, a.dtype # metadata only; nothing was downloaded +a[100:110, :50] # a NumPy array, fetched now +``` + +`https://` means a plain web server — nginx, a CDN, an S3 website endpoint — anything that answers a `Range` request. A Caterva2 subscriber is *not* reached that way: it names its datasets by root and path, so use {ref}`C2Array` (or `blosc2.URLPath` with {func}`blosc2.open`). + +## The cache + +Wrap either of those in a {ref}`Proxy` and what you read is kept: + +```python +p = blosc2.Proxy(b, urlpath="lung-cache.b2nd", mode="a") +p[10:12, 500:600] # fetched from the server, and written to the cache +p[10:12, 500:600] # read from the cache, no request at all +``` + +The cache is an ordinary Blosc2 file holding only the pieces you touched — a few hundred bytes for a freshly opened proxy over a 64 MB dataset. With `mode="a"` a later run picks up where the last one left off. `blosc2.open(url, lazy=True)` builds one for you; pass `cache_storage=` to say where it lives. + +## Only what a slice touches + +A chunk is the unit a container is compressed in, and it can be several megabytes. Fetching a whole one to read a corner of it is most of the cost of a remote read, so Blosc2 fetches **blocks** — the smaller pieces a chunk is built from — whenever a slice lands in a small part of a large chunk. + +You do not ask for this; it happens when it pays: + +- On S3, block reads are **5–17x faster** on arrays with multi-megabyte chunks, and **2–5x** on 1 MB ones. +- On cat2.cloud's `kevlar-tomo.b2nd`, a corner slice costs **0.031 MB instead of 2.723 MB**, and a slice touching ten chunks takes **0.14 s against 1.01 s**. + +It is never a loss. Two thresholds decide it — a chunk under a megabyte is one cheap request anyway, and wanting more than half a chunk's blocks is wanting the chunk — and both are answered from metadata already in hand. Where blocks are not available, the read falls back to whole chunks by itself: that happens for a dataset a Caterva2 subscriber *computes* rather than stores (a lazy expression, an HDF5 leaf, a `.b2z` member), and for a server that stops honouring ranges. + +Fetches also overlap: a lazy proxy runs 8 at a time by default. Pass `max_concurrency=1` for a local protocol with no latency to hide. + +## When the remote changes underneath + +A cache is only good while the bytes it was filled from are still there. Sources that can name their bytes — an fsspec URL by its token, a Caterva2 array by an identifier the subscriber keeps — are checked against what the cache recorded: + +```python +p = blosc2.Proxy(src, urlpath="cache.b2nd", mode="a") +# ValueError: the cache at cache.b2nd was built against different remote bytes; +# pass mode='w' to fetch them anew +``` + +`mode="w"` starts the cache empty and refetches. For a source that cannot name its bytes, the cache is adopted on geometry alone — same shape, dtype and partitioning — so an array rewritten in place while its geometry stayed the same is served from the cache as it was. Use `mode="w"` when that is a possibility. + +## Filling an array from several writers + +A Caterva2 array can be *written*, one chunk at a time, by as many processes as it has chunks. Lay the array out empty first — {func}`blosc2.uninit` writes a couple of hundred bytes whatever the shape — upload it to the subscriber, then have each writer post the chunks it owns: + +```python +import blosc2 +import numpy as np + +# Once, before the writers start: an empty array of the final geometry +blosc2.uninit( + (1_000_000,), + dtype=np.float64, + chunks=(100_000,), + blocks=(10_000,), + urlpath="run.b2nd", +) +``` + +Upload it with the client that comes with Caterva2: + +```sh +cat2-client upload run.b2nd @personal/run.b2nd +``` + +Then each writer opens it and posts its own chunks: + +```python +import math + +import blosc2 + +a = blosc2.C2Array("@personal/run.b2nd", urlbase="https://cat2.cloud/demo") +itemsize = a.dtype.itemsize +chunk = blosc2.compress2( + data, typesize=itemsize, blocksize=math.prod(a.blocks) * itemsize +) +a.update_chunk(nchunk, chunk) +``` + +Each slot is written once. A second write to the same slot raises {class}`blosc2.ChunkAlreadyWritten`, and that refusal is the whole of the coordination — two writers that both think they own a chunk are sorted out by the array, with no lease, lock or registry between them. The loser drops its chunk and moves on: + +```python +try: + a.update_chunk(nchunk, chunk) +except blosc2.ChunkAlreadyWritten: + pass # someone else got there first +``` + +Writing into an empty slot appends to the file and moves no other chunk, which is what makes a fill cheap and lets a reader follow one without its cached positions going wrong. {meth}`C2Array.written_chunks() ` says how far it has got, straight out of the file's own index — no endpoint of its own, about 2.5 ms over HTTP: + +```python +written = a.written_chunks() # one bool per chunk +print(f"{written.sum()}/{written.size} chunks in") +for nchunk in np.flatnonzero(~written): + ... # the work still to do, after a crash +``` + +What this buys: the subscriber serializes the writes themselves, so what overlaps is the round trip — which over a network is nearly all of the cost. Against a real subscriber, a fill went from **244 ms per chunk serially to 32 ms with 8 writers, 7.6x**. Over loopback, where there is no round trip to hide, it is 1.0x. + +## Your own transport + +If your frames live somewhere fsspec does not reach — per-request credentials, a signing proxy, a database column, an in-house gateway — supply one method and you get everything above: + +```python +import boto3 +import blosc2 + + +class S3Source(blosc2.ByteRangeNDSource): + def __init__(self, bucket, key): + self._s3 = boto3.client("s3") + self._bucket, self._key = bucket, key + self.stamp = self._s3.head_object(Bucket=bucket, Key=key)["ETag"] + super().__init__(f"s3://{bucket}/{key}") + + def read_range(self, offset, size): + answer = self._s3.get_object( + Bucket=self._bucket, + Key=self._key, + Range=f"bytes={offset}-{offset + size - 1}", + ) + return answer["Body"].read() + + +a = blosc2.Proxy(S3Source("bucket", "big.b2nd"), urlpath="cache.b2nd", mode="a") +``` + +(For plain S3 you would just use `blosc2.open("s3://bucket/big.b2nd", lazy=True)`; this is the shape of the thing.) + +Three things to get right: + +- **Set up the transport before `super().__init__()`.** The base constructor calls `read_range()` straight away to read the file's header. +- **`read_range()` must be thread-safe.** It is called from a thread pool so fetches can overlap. A boto3 *client* is fine; a `Session` or resource is not. +- **Set `stamp` if you can.** It is what lets a cache tell that the remote has changed. Without it the cache is kept on geometry alone. + +## See also + +- {doc}`Tutorial 6 <../getting_started/tutorials/06.remote_proxy>` — the same ground at a slower pace, with output. +- `examples/ndarray/rw-fsspec.py` — every way of reading and writing an fsspec URL, runnable. +- {ref}`C2Array`, {ref}`FsspecNDSource`, {ref}`ByteRangeNDSource`, {ref}`Proxy` — the reference pages. diff --git a/doc/reference/c2array.rst b/doc/reference/c2array.rst index da165991a..47d6a3ddb 100644 --- a/doc/reference/c2array.rst +++ b/doc/reference/c2array.rst @@ -14,6 +14,16 @@ HDF5 leaf) is fetched a whole chunk at a time, as everything was before. Which one this is takes at most one request to find out, and is decided once -- :meth:`C2Array.block_source` is what answers it. +A stored remote array can also be *filled*, by as many writers at once as it has +chunks. The array is laid out first -- ``blosc2.uninit`` writes a couple of +hundred bytes whatever its size -- and then each writer posts the chunks it owns +with :meth:`C2Array.update_chunk`. A slot nothing was written to is free, and a +write claims it; a second write to the same slot raises +:class:`blosc2.ChunkAlreadyWritten`, so two writers that both believe they own a +chunk are resolved by the array rather than by anything either of them holds. +:meth:`C2Array.written_chunks` reads how far the fill has got out of the frame's +own offsets, which is a couple of range reads and no endpoint of its own. + .. currentmodule:: blosc2 @@ -37,6 +47,9 @@ one this is takes at most one request to find out, and is decided once -- .. automethod:: __getitem__ +.. autoclass:: ChunkAlreadyWritten + + .. _C2NDSource: C2NDSource class diff --git a/plans/cat2-concurrent-writers.md b/plans/cat2-concurrent-writers.md new file mode 100644 index 000000000..1f05c07a0 --- /dev/null +++ b/plans/cat2-concurrent-writers.md @@ -0,0 +1,508 @@ +# Concurrent Chunk Writers For Caterva2 + +Written 2026-08-21, after [plans/cat2-block-granularity.md](cat2-block-granularity.md) +gave `C2Array` block-granular *reads* over HTTP ranges (branch +`cat2-block-granularity`; the Caterva2 side is `range-honesty`). + +Nothing here is implemented yet. Everything under *What was verified* was +measured or read out of the code on 2026-08-20/21; everything under *Plan* is +proposal. + +## The question + +Several processes, on several machines, want to fill one `.b2nd` living in a +Caterva2 server, each writing its own chunks, at the same time. What does that +take, and how much of the block-granularity work carries over? + +**Verdict: almost none of the read machinery carries over, and that is fine, +because the write path turns out to be small — one endpoint — provided the +array is pre-sized and each chunk is written exactly once.** The read side made +transports uniform behind one primitive (`read_range`); writing has no safe +mirror of that, and this stays a Caterva2 capability rather than a blosc2-remote +one (see [Non-goals](#non-goals)). + +## What was verified + +Measured on this machine (Apple M4 Pro, APFS, python-blosc2 4.11.1.dev0 against +c-blosc2 3.3.2), on local files. Code references: c-blosc2 at +`/Users/faltet/blosc/c-blosc2` (`main`), Caterva2 at +`/Users/faltet/ironArray/Caterva2` (`range-honesty`). + +### A contiguous frame compacts on rewrite, but appends on first write + +`frame_update_chunk()` (`blosc/frame.c:4281`) writes the new chunk *in place* and +moves the whole payload tail when the compressed size changes +(`frame.c:4546-4580`). Cost is therefore O(bytes after the chunk), not O(file), +and the physically-last chunk already costs nothing: `tail_nbytes` is 0 and the +move is skipped (`frame.c:4548-4549`). Median of 5 `update_chunk` calls, 1 MB +chunks, clevel 1: + +| file | chunk 0 | middle | last | same cbytes | +|---|---|---|---|---| +| 6.9 MB (20 chunks) | 1.482 ms | 0.852 ms | 0.244 ms | 0.285 ms | +| 27.6 MB (80) | 5.638 ms | 2.293 ms | 0.269 ms | 0.291 ms | +| 110.2 MB (320) | 21.213 ms | 10.828 ms | 0.296 ms | 0.282 ms | + +This is the behaviour of `9200990b` ("Fix contiguous-frame b2nd resize growth on +chunk updates", 2026-03-23), which replaced append-at-end-leaving-a-hole with +compaction. It is in the bundled 3.3.2. The older hole behaviour is not +recoverable as a flag without a format change: `get_coffsets()` locates the +offsets block at `header_len + cbytes` (`frame.c:1841-1846`), so the header's +`cbytes` is simultaneously the payload extent and the user-visible compressed +size; holes make those diverge and there is no second field. + +**But a chunk whose previous content was *special* does not compact at all.** +`old_chunk_is_regular = (!frame->sframe && old_offset >= 0)` (`frame.c:4377`); +zero/NaN/uninit chunks live entirely in the offsets with the high bit set and +carry no payload, so there is no tail to move and the new chunk is appended at +`new_chunk_offset = cbytes` (`frame.c:4379`). Filling a 320-chunk array +pre-sized with `blosc2.uninit()`, in random chunk order: + +``` +pre-sized uninit file on disk: 221 bytes (335.5 MB logical) +fill all 320, random order: median 0.459 ms/chunk, max 4.136 ms → 110.2 MB +rewrite an already-written one: 9.157 – 14.029 ms (the compaction above) +``` + +Flat, position-independent, and **no other chunk's offset changes**. That one +fact is what the whole design below is built on. + +(`clevel=0` gives the same flatness for repeated rewrites, since every chunk is +exactly `nbytes + overhead` and the tail never moves — 0.44-0.49 ms at any +position on a 336 MB file. Kept here as a note; the write-once design does not +need it.) + +### `uninit` is a usable sentinel; `zeros` is not + +Both are special chunks, so "written or not" is legible from the offsets in +either case — the tag carries it, never the data. The difference is what +happens when a writer legitimately stores an all-zero chunk: + +``` +compress2(np.zeros(...)) → cbytes=32, special=ZERO +compress2(np.arange(...)) → cbytes=1152, special=regular +``` + +Blosc2 detects the run and emits a special ZERO chunk, so with a `zeros` +pre-fill a genuinely-all-zero written chunk is indistinguishable from a +never-written slot. With `uninit` the two separate cleanly +(`schunk.iterchunks_info()`): + +``` +chunk 0: special=ZERO ← written, data really was zeros +chunk 1: special=NOT_SPECIAL ← written, real data +chunk 2: special=UNINIT ← never written +``` + +Cost of `uninit`: an unwritten chunk reads as undefined bytes, so completeness +has to be part of the contract rather than a nicety. See +[Progress is the offsets block](#progress-is-the-offsets-block). + +### The frame length is not a validator + +A special-chunk write sets `chunk_cbytes = 0` and leaves `new_cbytes` +unchanged, so the file length moves only if the recompressed offsets block +happens to change size: + +``` +after uninit create size= 221 md5=672911ce0aca +after ZERO chunk write size= 277 md5=063520e5ea6c +after 2nd ZERO write size= 277 md5=63bf170ef1c8 ← same length, new content +after regular write size= 1429 md5=9aed1d5dbddb +``` + +Worse than a missed invalidation: since `new_cbytes == cbytes`, that write +rewrites the offsets block **in place**, where a regular append writes it past +the new chunk. So a zeros write both opens a torn-read window on the offsets +and is invisible to a length check. + +### The generation counter is + +`.b2lock` carries a `uint64` at offset 8 (`FRAME_LOCK_SEQ_OFFSET`, +`frame.c:130`), bumped by every exclusive acquisition (`frame.c:269-271`). +c-blosc2's own comment states the reason: it "detects mutations by other handles +exactly, even when the frame length on disk ends up unchanged". It lives +outside the frame bytes, so only a server with local filesystem access can serve +it — which is exactly what Caterva2 is. + +### Caterva2 has no chunk-write endpoint, and its write path is accidentally safe + +Write surface today is `api/upload` (whole file, `server.py:1279`), `api/append` +(axis 0, `server.py:1413`) and `api/upload_lazyarr`; `api/chunk` +(`server.py:924`) is GET-only. Neither write endpoint takes any lock. They are +safe today only because they are `async def` bodies that never await across +their blocking blosc2 calls, in a single-process deployment +(`uvicorn.run(app)`, `server.py:3351`). Moving the write to a threadpool — +which concurrency requires — removes that accident, so the locking is not +optional extra credit. + +`locking=True` (`src/blosc2/storage.py:212`), `holding_lock()` +(`src/blosc2/schunk.py:476`) and the cross-process multi-writer tests already +exist; see `todo/locking-mwmr.md`, whose item 7 is this use case. + +### Pre-sizing needs no new endpoint + +A pre-sized uninit array is **221 bytes for a 335.5 MB logical array**, and +`.b2nd` is in `BLOSC2_NATIVE_SUFFIXES` (`caterva2/services/srv_utils.py:35`), so +`api/upload` already stores it verbatim. Creation is +`blosc2.uninit(...)` locally plus an existing upload; the file *is* the geometry +specification. (Quota is then accounted at 221 bytes, so the chunk-write +endpoint has to re-check it — see phase 1.) + +## The design + +### Pre-sized, write-once + +1. The owner creates the array locally with `blosc2.uninit(shape, dtype, chunks, + blocks, cparams)` and uploads it (~200 bytes). Geometry is fixed here and + never changes: **no writer ever resizes**. +2. Writers own disjoint chunk indices, agreed between themselves; the server + does not arbitrate the partition. +3. Each chunk is written **exactly once**. A second write is refused. + +Everything good follows from 3: writes never move data (~0.5 ms), never +invalidate another reader's chunk offsets, and never need a read-modify-write of +a partially covered chunk. + +### Progress is the offsets block + +The UNINIT-vs-everything-else tag *is* the completion record. No manifest, no +sidecar bitmap, no progress endpoint: + +- **Write-once enforcement**: the server checks slot *n* is UNINIT before + accepting. Note it must test UNINIT specifically, not "is special" — a + written all-zero chunk is special too. +- **Atomic by construction**: the tag flips in the same offsets rewrite that + publishes the chunk, under the same lock. No window where a chunk is on disk + but unrecorded, or the reverse. +- **Readers get it free**: `ByteRangeNDSource` already decodes a negative offset + and reconstructs the special chunk locally (`src/blosc2/proxy_source.py:706`, + `853`), so an unwritten chunk costs zero bytes and zero requests. +- **Progress is a couple of range reads**: the offsets block is a single span + the branch already knows how to locate -- through the frame's header, which a + write moves too, so following a fill re-reads that first. + +It deliberately records no in-progress state, no identity, no timing and no +history. That gives crash *recovery* (rerun the unwritten set) but not +*leases*: two writers who both believe they own chunk 7 are resolved by the +refusal, not prevented. + +### The one remaining tearing window + +For a regular append, the new chunk is written at `header_len + cbytes` — which +is exactly where the *old* offsets block lives. A reader that fetched the +header and then reads the offsets can therefore land on a half-written chunk. +This is the branch's two-request open, and it is why an ETag is load-bearing +rather than a nicety. + +## Plan + +### Phase 1 — `POST api/chunk/{path}` (caterva2) — the main piece + +Body is one compressed chunk; `nchunk` is a query parameter. Under +`get_writable_path(path, user)`: + +1. Refuse anything not a stored contiguous `.b2nd` — lazy expressions, `.b2z` + members, HDF5 leaves. `api/info`'s discriminator from phase 2 of the + block-granularity plan already reasons about this. +2. Validate the chunk header against the array's geometry (`nbytes`, + `blocksize`, `typesize`). A mismatched chunk corrupts the array outright, so + this is not optional. +3. Re-check quota against the *delta*, since creation only accounted ~200 bytes. +4. Open with `locking=True`, and inside `holding_lock()`: read the offsets, + refuse with **409** unless slot *n* is UNINIT, then `update_chunk`. +5. Run the whole thing in a threadpool — it is blocking, and it must not hold + the event loop. + +Acceptance: N processes filling disjoint chunk sets of one array converge to the +exact expected contents; a second write to any slot returns 409; a torn or +mis-shaped chunk is refused before it reaches `update_chunk`. + +### Phase 2 — ETag from the generation counter (caterva2) — small, load-bearing + +Serve the `.b2lock` counter as a strong `ETag` on `api/info` and on ranged +`api/fetch`/`api/download`, so a client can prove its header and its offsets came +from the same frame. A `pread` of 8 bytes. + +- Not the file length: proved above that a zeros write can leave it unchanged. +- Not `If-Match` on the *write* path: the UNINIT check is already the + compare-and-swap, and a better one — it tests the real state, not a token. +- Define the fallback for an array with no sidecar yet (never written under + locking): either create it on first open, or serve a documented weaker + validator. + +### Phase 3 — `C2Array.update_chunk` / `written_chunks` (blosc2) — small + +`update_chunk(nchunk, chunk)` and `aupdate_chunk` through the pooled client the +branch added, plus `written_chunks() -> np.ndarray[bool]`: the frame's header +and then the offsets it locates, decoded locally. No general `__setitem__`: a partially covered chunk +is a networked read-modify-write and would need CAS to be safe. + +### Phase 4 — `stamp`: which array, and has it changed (blosc2) — small + +`C2Array.stamp` is `mtime:cbytes` (`src/blosc2/c2array.py:749`) and answers "are +these the same bytes?". Under append-only writing the answer is "no" after +every chunk write, which would discard a `Proxy` cache that is still entirely +valid, since existing chunks never move. The stamp needs to answer the narrower +question — *replaced, or merely appended to?* There is no UUID in the frame +header, so this is the one genuinely open design question here. Options to +weigh: a server-side identity token (inode + creation time) carried in +`api/info`; a creation nonce written into vlmeta at pre-size time; or splitting +the stamp into an identity part and a freshness part. + +### Phase 5 — Completion and publish (caterva2) + +The completion condition is free: after each accepted write, inside the same +`holding_lock()` region, scan the already-decompressed offsets for remaining +UNINIT slots. State lives in vlmeta: `filling → publishing → published(url)`. + +- On zero remaining, compare-and-set `filling → publishing`. The lock makes it + **exactly-once**: two writers finishing together both see zero, one wins the + flip, the winner owns the publish. +- Do the upload **outside** the lock, then flip to `published` with the URL. A + slow upload must not block writers. +- `POST api/publish/{path}` is the primitive; auto-trigger on completion is a + thin layer over it, which also gives a manual retry for the stuck-in- + `publishing` case. +- **The destination must not come from the client.** A client-supplied `s3://` + URL lets the server be aimed at a bucket the caller controls. The server + config names the destination root; the array supplies a relative key only. + Credentials stay on the server, which also means writers never hold them. + +What lands in S3 is a finished contiguous frame — exactly what this branch's +`FsspecNDSource` reads with byte ranges. Caterva2 is the write path, the object +store is the read path, and both ends already work. Publishing has none of the +problems of writing chunks to S3: the frame is immutable by then, so no locking, +no ETag, no partial writes. + +Acceptance: an array filled by N writers publishes exactly once, is readable +from S3 by `blosc2.open(url, lazy=True)` with block granularity, and a crash +mid-publish is recoverable through the explicit endpoint. + +### Phase 6 — Tests and a bench + +- Cross-process hammer: N writers × disjoint chunks against a live server, plus + a reader sampling throughout; assert no torn chunk, exact final contents, and + `written_chunks()` monotone. +- The zeros case explicitly: an array whose writers all send ZERO chunks must + still complete and publish (this is the case `zeros` pre-filling would break). +- ETag: a zeros write must change it (the length does not). +- Extend `bench/ndarray/cat2-block-granularity.py`'s stand-in server to accept + chunk writes, so the write path is measurable without a deployment, the way + the read path already is. + +## Risks and open questions + +- **Phase 4 is unresolved** and everything else can land without it; the cost of + deferring is that a `Proxy` over an array still being filled re-fetches more + than it needs. +- **Crash mid-fill** leaves an array permanently incomplete. Correct, but a + coordinator needs `written_chunks()` and a reassignment story; consider a + reporting-only staleness timeout. +- **Multi-worker deployment**: `locking=True` covers the frame, but the + process-local caches in `server.py` (the mtime-keyed opened-array cache, the + `locks` dict at `server.py:494`/`952`) do not. Decide whether multi-worker is + in scope now or after. +- **Crash mid-write** hands the next lock holder a possibly torn frame; there is + no journal. Same accepted limitation as item 5 of `todo/locking-mwmr.md`. +- **Lock fairness**: `flock` has no FIFO ordering, so a read-heavy array could + starve writers. + +## Non-goals + +- **Chunk writes over fsspec/S3.** Three independent blockers, only the last of + which is about validators: object stores have no partial write, so every chunk + write rewrites the whole frame object; the offsets block is shared mutable + state, so concurrent writers lose updates (S3 conditional writes give CAS, but + each retry is another full-object rewrite, so it degrades exactly where it + should scale); and there is no lock, hence no generation counter. The read + side generalised because reading needs one primitive that every backend has. + Writing needs mutual exclusion plus partial in-place writes, and only a server + with a real filesystem has both. Do not add a `write_range` to + `ByteRangeNDSource`. +- **Rewriting live chunks.** Allowed in principle, costs 9-21 ms of compaction, + and forfeits write-once enforcement, offset stability and the completion + record all at once. Refuse it; revisit only with a use case. +- **Leases / ownership arbitration.** The offsets record what is written, not + who is writing. An external coordinator's job. +- **Server-mediated writes to a frame that lives on S3.** Interesting, and the + natural extension of phase 5 in the other direction; out of scope here. +- **A hole-plus-repack update mode** to make rewrites cheap. Needs a second + counter in the format (the trailer is msgpack and variable, so a softer home + than the fixed header) plus a `vacuum`/`repack`, which neither repo has. + Format project, not a plan item. + +## Reproducing the measurements + +Each was a short script run against local files; none needs a server. + +- **Rewrite cost by position**: build a contiguous `.b2nd` of *n* 1 MB chunks, + time `schunk.update_chunk()` on chunk 0, *n*/2 and *n*-1 with freshly + compressed data, and again with the chunk's own bytes (the same-cbytes case). +- **Append on a special slot**: `blosc2.uninit(...)`, then fill all chunks in a + random permutation, timing each; then rewrite three of them. +- **Sentinel**: `compress2` an all-zero buffer and read the special bits at + `chunk[31] >> 4 & 0x7`; cross-check with `schunk.iterchunks_info()`. +- **Length is not a validator**: stat + md5 the file after a create, two ZERO + writes and a regular write. + +## What landed + +All of it (2026-08-21), on `cat2-concurrent-writers` here and `c2cache-monorepo` +in Caterva2. Phase 4 landed last and is written up separately below, because +what it found changed what it should do. + +| phase | where | commit | +|---|---|---| +| 1. `POST api/chunk` | caterva2 `server.py` | *Accept one chunk at a time into a slot that holds none* | +| 2. ETag | caterva2 `server.py` | the same commit | +| 3. client writes | blosc2 `c2array.py`, `proxy_source.py` | *Write a chunk of a remote array, and read which ones were written* | +| 5. completion and publish | caterva2 `server.py` | *Publish an array once every one of its chunks has landed* | +| 5a. atomic publish | caterva2 `server.py` | *Move a published array into place instead of streaming into it* | +| 6. the bench | blosc2 `bench/ndarray/` | *Measure a fill the way the reads are measured* | +| 4. the stamp | both | *Name a filled array by a nonce, and say when it is complete* | + +The shape held: a pre-sized `uninit` array, one write per slot, the offsets as +the record, and a 409 as the whole of the coordination. 15 tests against a live +subscriber (`caterva2/tests/test_chunk_writes.py`) and 10 against a stand-in +(`tests/ndarray/test_c2array_writes.py`), which is where the client-side +behaviour is pinned without a service. + +### What the work found + +- **Reading the index is not the same question as reading blocks.** + `C2Array.written_chunks()` was gated on `serves_blocks`, which also weighs + whether *splitting a chunk into blocks* would pay — a frame of small chunks + reported that it served no blocks and so could not say which chunks were + written either. The geometry half is now `_reports_geometry` and the index is + read whatever the chunks cost. `serves_blocks` is unchanged for the block + path, which reads it at `Proxy` build time and must keep costing nothing. +- **Invalidating the index means the header too.** A write moves the frame's + length and its payload extent, and the offsets are found through both, so + dropping the offsets alone left the next read looking for them at the old + position. `ByteRangeNDSource.invalidate_index()` marks both stale and reads + neither until something asks. +- **The completion scan had to move off `iterchunks_info`.** It reads a lazy + chunk apiece — 3.5 µs each, so 17.8 ms on 5000 chunks, which would have made a + fill cost the square of its length. The write-once check reads the one + chunk's header instead (~3 µs), and the count comes from the offsets in one go + (0.37 ms at 5000 chunks, 0.39 ms at 20000 — flat where the walk is linear). +- **A publish that streams into place is readable before it is whole.** The + destination file exists from its first byte and a frame is not readable until + its last, so a reader polling for the published array opened it mid-copy and + got a NULL back. Found by the test doing exactly that, which had passed only + while the copy won the race. Published under a name of its own and moved into + place. +- **A stale blosc2 handle corrupts a write silently.** Two handles open over one + frame, one of them writing, leaves the frame unreadable — `Invalid arguments + for stdio write` under `BLOSC_TRACE=1`, and nothing at all without it: the + write itself does not raise. This is the hazard `todo/locking-mwmr.md` + documents, but the silence of it is worth knowing. Both the endpoint and the + test stand-in are written to hold exactly one handle and to drop it before + anything reads the file again. + +### Phase 4, decided (2026-08-21) + +Settled with the vlmeta nonce, but **not** the way this plan first framed it. +The framing was wrong, and measuring it is what showed that. + +The complaint above was that a cache over an array being filled is discarded on +every write "though every chunk it holds is still exactly where it was". That is +true of the chunks that were *written* when the cache was built, and false of the +ones that were not. A `Proxy` reading a slice of an unwritten chunk caches the +zeros an unwritten chunk reads as, and caches its run-length offset with them; +when a writer fills that slot, both are wrong and nothing in the cache marks them +apart from the chunks that are still good. Pinned by giving a source a stamp +that never moves and watching it happen: + +``` +read while unwritten: [0 0 0] +the file now holds: [7 7 7] +what the cache serves: [0 0 0] <- stale, and silent +``` + +So the stamp of an array still being filled *must* keep moving on every write. +That is not waste; it is the only correct answer. What is worth fixing is the +other two things: + +- **Which array is this.** `mtime:cbytes` can be repeated by a different array + that came to sit at the same path — two arrays of constant chunks compress to + the same size, and an mtime can be set. A cache of the first served against + the second is wrong in every chunk and says nothing. The subscriber now writes + a nonce into vlmeta the first time a chunk lands, and `api/info` already + carries vlmeta, so reading it costs no request. +- **When it stops changing.** Every slot of a complete array is claimed, so + every write to it is refused and its bytes cannot move again. The subscriber + records that (`fill_state` leaves `filling` on the last chunk, whether or not + there is anywhere to publish to), and a complete array is then stamped by its + nonce and its size alone — so a cache of it survives an mtime that churned for + reasons of its own, which is what a republish or a copy does. + +| the array | stamp | +|---|---| +| being filled | `n::` — moves on every write, and must | +| complete | `n:` — holds still, and may | +| never filled chunk-wise | `:` — exactly as before | + +The finished array is the one read again and again, so that is where the win is. +Measured on the stand-in: a cache of a complete array, reopened after its mtime +moved, refetches **nothing**. With the old stamp the same reopen raises +`the cache ... was built against different remote bytes` and the whole cache is +thrown away. + +What the nonce does not do: it names the array's lineage, not its bytes. Someone +who uploads an edited copy of a complete array, vlmeta and all, is served the old +cache. Nothing short of a content hash closes that, `mtime:cbytes` did not close +it either, and a write-once array has no ordinary path to it. + +### The write path, measured + +`bench/ndarray/cat2-block-granularity.py --write` (2026-08-21), which lays out an +empty array of the dataset's geometry, fills it with the dataset's own chunks, +and times the three things the design rests on. Against the stand-in over +loopback with a WAN put in front (`--latency-ms 45 --bandwidth-mbs 10`), 8 chunks +of 1.76 MB: + +| | | | +|---|---|---| +| 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 | the header, then the offsets | +| the same, local | 0.33 ms | one decompress, whatever the count | +| `iterchunks_info()`, local | 9.7 µs **per chunk** | what grows with the array | + +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 +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. + +Also verified end to end against a real `cat2-server`: six chunks filled through +the endpoint at 5.7 ms each over localhost, read back identical to the source, +and the unwritten remainder reading as undefined bytes — which is what makes the +completeness contract part of the API rather than a nicety. + +### The way in, from the Caterva2 client + +The endpoints existed and nothing but `C2Array` could reach them, so the +Caterva2 client described uploading and appending and said nothing about the one +way to write an array from several processes at once. `Client.lay_out`, +`fill_chunk`, `written_chunks` and `publish` are the four calls the workflow is +made of, with `Dataset` methods to match; the two that move chunks delegate to +`C2Array` rather than reimplementing the request, so the refusal a second write +earns is raised in one place only. `publish_root` reaches +`caterva2-server.sample.toml` too, which was the only way anyone deploying would +find out that filled arrays can be published at all. + +An array laid out this way is measured at **under 4 KB for a shape of 20 GB**, +which is the property the whole arrangement leans on: an unwritten chunk lives +in the frame's offsets and nowhere else. + +### Left undone +- Multi-worker deployment. `locking=True` covers the frame across processes and + the `.b2lock` counter is read from disk, so the ETag is right there too; the + per-path `asyncio.Lock` is not, and neither are the mtime-keyed open-array + caches. Nothing here depends on it, and nothing here provides it. diff --git a/src/blosc2/__init__.py b/src/blosc2/__init__.py index 14433aac1..182111bc6 100644 --- a/src/blosc2/__init__.py +++ b/src/blosc2/__init__.py @@ -572,7 +572,7 @@ def _raise(exc): from .ref import Ref from .b2objects import open_b2object -from .c2array import c2context, C2Array, C2NDSource, URLPath +from .c2array import c2context, C2Array, C2NDSource, ChunkAlreadyWritten, URLPath from .dsl_kernel import DSLSyntaxError, DSLKernel, dsl_kernel, validate_dsl, validate_dsl_jit from .lazyexpr import ( @@ -859,6 +859,7 @@ def _raise(exc): # Classes "C2Array", "C2NDSource", + "ChunkAlreadyWritten", "Column", "CParams", "CTable", diff --git a/src/blosc2/c2array.py b/src/blosc2/c2array.py index 4a160980c..beb9f45ba 100644 --- a/src/blosc2/c2array.py +++ b/src/blosc2/c2array.py @@ -7,6 +7,7 @@ from __future__ import annotations +import asyncio import atexit import math import os @@ -208,6 +209,43 @@ def _xpost(url, json=None, auth_token=None, timeout=TIMEOUT): return response.json() +def _chunk_headers(auth_token): + """What a chunk write is sent with: bytes, not the JSON `_xpost` sends.""" + return _auth_headers(auth_token, {"Content-Type": "application/octet-stream"}) + + +def _chunk_written(response, url, nchunk): + """Read a chunk write's answer, in one place for both ways of sending it. + + The write contract lives here rather than at each call site: a slot that was + already claimed is the one refusal a writer is meant to act on, and it must + read the same whether the request went out on the pooled client or on the + async one. + """ + if response.status_code == 409: + raise ChunkAlreadyWritten(f"{url} already holds a chunk at {nchunk}") + response.raise_for_status() + return response.json() + + +def _xpost_bytes(url, content, params=None, auth_token=None, timeout=TIMEOUT): + """POST a body of bytes through the pooled client, and read what came back. + + `_xpost` sends JSON, which a compressed chunk is not: it goes as it is, and + the subscriber reads it as the chunk it will store. + """ + response = _sync_client().post( + url, params=params, content=content, headers=_chunk_headers(auth_token), timeout=timeout + ) + return _chunk_written(response, url, params and params.get("nchunk")) + + +async def _axpost_bytes(client, url, content, params=None, auth_token=None): + """The same request off the event loop; see :func:`_xpost_bytes`.""" + response = await client.post(url, params=params, content=content, headers=_chunk_headers(auth_token)) + return _chunk_written(response, url, params and params.get("nchunk")) + + def _sub_url(urlbase, path): urlbase = urlbase or _subscriber_data["urlbase"] if not urlbase: @@ -383,6 +421,17 @@ def _span_of(parts: list[tuple[int, bytes, int | None]], offset: int, size: int, raise PartsMissing(f"{url} answered without the bytes at {offset}, which were asked for") +class ChunkAlreadyWritten(ValueError): + """A chunk was written to a slot of a remote array that already held content. + + A subscriber that accepts chunk writes accepts each slot exactly once: the + frame's own offsets say whether a slot was ever written, and a second write + would move every chunk that came after it. So a writer that finds this has + lost a race, or is repeating work another writer already did; either way the + array is intact and the chunk it carried is the one to drop. + """ + + class C2NDSource(ByteRangeNDSource): """The frame behind a :ref:`C2Array`, read over HTTP byte ranges. @@ -540,7 +589,14 @@ 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() - # An index a `Proxy` handed over before the source existed; see adopt_index + # Set when this handle writes: `meta` describes the array as it was read, + # and a write of its own moves everything `api/info` reports about it. + # The epoch counts those writes, so a read of `api/info` that was in + # flight when one landed can tell that its answer predates it + self._meta_stale = False + self._meta_epoch = 0 + self._meta_lock = threading.Lock() + # An index a `Proxy` handed over before the source existed; see _adopt_index self._pending_index = None # Try to 'open' the remote path @@ -700,7 +756,7 @@ def get_chunk(self, nchunk: int) -> bytes: 23., 27., 28., 10., 11., 0., 0., 30., 31., 0., 0., 12., 13., 0., 0., 32., 33., 0., 0.], dtype=float32) """ - url = _sub_url(self.urlbase, f"api/chunk/{self.path}") + url = self._chunk_url() params = {"nchunk": nchunk} response = _xget(url, params=params, auth_token=self.auth_token) return response.content @@ -726,7 +782,7 @@ async def aget_chunk(self, nchunk: int) -> bytes: out: bytes The requested compressed chunk. """ - url = _sub_url(self.urlbase, f"api/chunk/{self.path}") + url = self._chunk_url() params = {"nchunk": nchunk} headers = _auth_headers(self.auth_token) if self._aclient is None: @@ -741,6 +797,202 @@ async def aclose(self) -> None: await self._aclient.aclose() self._aclient = None + # -- Writing chunks. A pre-sized array is filled a chunk at a time, by as + # many writers as there are chunks to fill; the subscriber serializes them + # and refuses a slot that was already written. + + def update_chunk(self, nchunk: int, chunk: bytes) -> dict: + """Write one compressed chunk into a slot of the remote array. + + The array has to exist and to be laid out already -- `blosc2.uninit` and + an upload is what makes one -- and the slot has to be one nothing was + ever written to. That is not a restriction the transport invents: a + chunk written into an empty slot is appended to the frame and moves + nothing, while one written over a chunk that is already there moves every + byte after it, so a fill made of writes-once is the cheap one and the one + whose offsets a concurrent reader can keep. + + The chunk must match the array's geometry -- its chunkshape, its typesize + and its blocksize -- which is what compressing against + :attr:`cparams` and :attr:`blocks` gives; the subscriber checks it and + refuses anything else rather than storing a chunk the array cannot read. + + Parameters + ---------- + nchunk: int + Which chunk of the array to write, numbered as + :meth:`NDArray.get_chunk` numbers them. + chunk: bytes + The compressed chunk, as :meth:`SChunk.get_chunk` or + :func:`blosc2.compress2` produce it. + + Returns + ------- + out: dict + What the subscriber reports of the array's state now. Carries + ``written`` and ``nchunks`` where it counts them, so a writer can see + a fill finish without asking again. + + Raises + ------ + ChunkAlreadyWritten + The slot already holds a chunk. The array is untouched. + + Examples + -------- + >>> import math, blosc2, numpy as np # doctest: +SKIP + >>> a = blosc2.C2Array("@personal/run.b2nd", urlbase) # doctest: +SKIP + >>> data = np.arange(math.prod(a.chunks), dtype=a.dtype).reshape(a.chunks) # doctest: +SKIP + >>> itemsize = a.dtype.itemsize # doctest: +SKIP + >>> chunk = blosc2.compress2( # doctest: +SKIP + ... data, typesize=itemsize, blocksize=math.prod(a.blocks) * itemsize + ... ) + >>> a.update_chunk(0, chunk) # doctest: +SKIP + {'written': 1, 'nchunks': 320} + + The blocksize is spelled out because :func:`blosc2.compress2` picks its + own when it is not: left to choose it takes the whole chunk, and a chunk + blocked differently from the array is one the subscriber refuses. + """ + url = self._chunk_url() + try: + return _xpost_bytes(url, chunk, params={"nchunk": nchunk}, auth_token=self.auth_token) + finally: + # However it went. A refusal is the answer of a subscriber that has + # already stored someone else's chunk in that slot, and a request that + # failed on the way home may have stored this one; either way what + # this handle read of the array is no longer what the array is + self._forget_index() + + async def aupdate_chunk(self, nchunk: int, chunk: bytes) -> dict: + """Write one compressed chunk asynchronously; see :meth:`update_chunk`. + + The same request, off the event loop, so a writer with many chunks to + send can have several in flight. The subscriber serializes them at the + far end regardless -- what overlaps is the round trip, which for a + chunk-sized body is most of the cost. + """ + url = self._chunk_url() + if self._aclient is None: + self._aclient = _httpx().AsyncClient(timeout=TIMEOUT) + try: + return await _axpost_bytes( + self._aclient, url, chunk, params={"nchunk": nchunk}, auth_token=self.auth_token + ) + finally: + # Off the loop as well: `_forget_index` waits on the lock a source + # being opened holds, and that open is a request of its own -- parking + # the loop on it would stall every write still in flight, which is the + # whole of what this method has over the blocking one + await asyncio.to_thread(self._forget_index) + + def written_chunks(self) -> np.ndarray: + """Which chunks of the remote array hold content; see + :meth:`ByteRangeNDSource.written_chunks`. + + Read out of the frame's own offsets, which is where a fill records + itself: no endpoint of its own, and nothing for the subscriber to keep in + step with the array. Read afresh every time, since the point of asking + is to see what other writers have done since -- which is a couple of + range reads, the header first (a write moves the frame's length, and the + offsets are found through it) and then the offsets it locates. + + Nothing else about the handle is disturbed: this asks what the *array* + holds, not what this handle has done, so `meta` is left as it was and no + `api/info` is spent on it. + """ + with self._ranged(index_only=True) as source: + # Through the source rather than around it: `_ranged` is what builds + # one, and a source built here takes up any index a `Proxy` left in + # `_pending_index` -- which is as old as the cache it came from. + # Invalidating what has just been built is what makes this a read of + # the frame rather than of whatever was already believed about it + source.invalidate_index() + return source.written_chunks() + + def _chunk_url(self) -> str: + """Where a chunk of this array is read from, and written to.""" + return _sub_url(self.urlbase, f"api/chunk/{self.path}") + + def _forget_index(self) -> None: + """Drop what this handle read of a frame it has since written to.""" + # 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. Read again when something + # asks, rather than here, so a writer that never asks pays no request + with self._meta_lock: + self._meta_stale = True + self._meta_epoch += 1 + 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() + # And an index that never reached a source: it came out of a `Proxy` + # cache filled before this write, so a source built later must not + # start from it + self._pending_index = None + + def _reread_meta(self) -> None: + """Read `api/info` again, and keep the answer if it is still an answer. + + The request is made outside the lock -- it is a round trip, and holding a + lock across one would serialize every reader of this handle behind it -- + so a write of this handle's can land while it is in flight. Such an + answer describes the array as it was before that write: it is dropped, + and the handle left marked stale, rather than stored as current and the + write it predates forgotten along with it. + """ + with self._meta_lock: + seen = self._meta_epoch + meta = info(self.path, self.urlbase, auth_token=self.auth_token) + with self._meta_lock: + if self._meta_epoch != seen: + return + self.meta = meta + self._meta_stale = False + + def _refresh_meta(self) -> None: + """Read `api/info` again, if this handle has written since it last did. + + Every property built on `meta` goes through this, so that what they say + does not depend on which of them was read first. It costs nothing to a + handle that has not written -- which is every reader -- and one request + to one that has. + """ + if self._meta_stale: + self._reread_meta() + + @property + def _meta_complete(self) -> bool: + """Whether `meta` describes an array that can no longer change. + + Every slot of a filled array is claimed, so every write to it is refused: + what `api/info` says of one is what it will go on saying. Anything else + -- an array still being filled, or one that was never filled a chunk at a + time and so says nothing either way -- can move under this handle at any + moment, and asking again is the only way to find out. + """ + vlmeta = self.meta.get("schunk", {}).get("vlmeta") or {} + return vlmeta.get("fill_nonce") is not None and vlmeta.get("fill_state", "filling") != "filling" + + def refresh_stamp(self) -> None: + """Look at the array again, so that :attr:`stamp` speaks for it now. + + `meta` is read when the handle is opened and, of itself, never again: a + `stamp` off it names the array as this handle last saw it, which for a + handle that has outlived someone else's writes is not the array. A + `Proxy` calls this before it reads the stamp it will judge its cache by, + which is the one moment that difference decides anything. + + One `api/info`, and none at all for an array already known to be complete + -- nothing can write to one of those, so nothing it reports can move. + """ + if self._meta_stale or not self._meta_complete: + self._reread_meta() + # -- 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 # round; every one of them falls back to `get_chunk` when it is not. @@ -753,16 +1005,51 @@ def stamp(self) -> str | None: filled from: a shape and a partitioning survive a rewrite, while every cached chunk -- and, in block mode, every offset they were fetched by -- goes stale. The subscriber's own mtime does tell, and `api/info` carries - it, so this costs no request; the compressed size goes in with it, since - a rewrite within the same clock tick is what an mtime cannot see. - - None when the subscriber reports no mtime, which leaves the cache checked - on its geometry alone, as every source without a stamp is. + it, so this costs no request of its own; the compressed size goes in with + it, since a rewrite within the same clock tick is what an mtime cannot + see. What it names is the array as this handle last looked at it -- + :meth:`refresh_stamp` is how a caller that needs it to be the array *now* + says so, and what a `Proxy` calls before judging a cache by it. + + Two questions, and they want different answers. *Which array is this* is + answered by the nonce a subscriber writes into an array's vlmeta the first + time a chunk is written to it: a size and an mtime can both be repeated + by a different array that came to sit at the same path, and a cache + served against one of those is stale without ever saying so. *Has it + changed since* is answered by the mtime and the compressed size, as + before. + + The second question stops being worth asking once the array is complete. + Every slot of a filled array is claimed, so every write to it is refused, + and the bytes a cache holds cannot move again -- so a complete array is + stamped by its nonce and its size, and a cache of it survives an mtime + that churned for reasons of its own. + + An array still being filled is stamped freshly on every write, and has to + be. A cache built while a chunk was unwritten holds that chunk as the + zeros an unwritten chunk reads as, and holds its offset as the run-length + one it had; when a writer fills that slot, both are wrong, and nothing in + the cache marks them apart from the chunks that are still good. + + None when the subscriber reports no mtime and the array carries no nonce, + 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", "") mtime = self.meta.get("mtime") - if mtime is None: - return None - return f"{mtime}:{self.meta['schunk'].get('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: @@ -787,12 +1074,19 @@ def serves_blocks(self) -> bool: blosc2 declines to split a chunk below ``BLOCK_MIN_CBYTES``, so the block path would end in whole chunks anyway, by the longer road. + Read off `api/info` again where this handle has written since it last + looked, which is the one case where the answer moves under it: a pre-sized + array holds almost nothing until it is filled, and a writer that took the + open-time figure would go on calling its own filled array too small to + take apart. That is one request to a handle that has just written, and + none at all to a reader -- which is what the promise below needs. + False is the whole answer; True is only that it is worth one request to find out, which :meth:`block_source` spends. A :ref:`Proxy` reads this when it is built, to decide whether its cache records blocks or chunks, so it must cost nothing and must not depend on what has been fetched. """ - if not all(key in self.meta for key in ("chunks", "blocks", "schunk")): + if not self._reports_geometry: return False try: nchunks = math.prod(math.ceil(s / c) for s, c in zip(self.shape, self.chunks, strict=True)) @@ -802,6 +1096,18 @@ def serves_blocks(self) -> bool: # whole chunks work for, as they do for every dataset there is return False + @property + def _reports_geometry(self) -> bool: + """Whether `api/info` describes a stored dataset rather than a computed one. + + Necessary for reading the frame at all, where :attr:`serves_blocks` is + that plus a judgement about whether taking its chunks apart would pay. + The frame's own index is worth reading either way: it is a range read or + two, and it is what says where the chunks are and which were written. + """ + self._refresh_meta() + return all(key in self.meta for key in ("chunks", "blocks", "schunk")) + def block_source(self) -> C2NDSource | None: """The frame reader behind the block methods, or None if there is none. @@ -809,7 +1115,27 @@ 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() if self.serves_blocks else None + + def _index_source(self) -> C2NDSource | None: + """The same reader, built for any stored frame however small its chunks. + + Reading the frame's index is not the same question as reading blocks of + its chunks: a frame of chunks too small to take apart still has offsets, + 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() if self._reports_geometry else 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: @@ -825,16 +1151,17 @@ def _open_block_source(self): 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 - if not self.serves_blocks: + # 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 # `.b2z` member reports one and is streamed all the same try: source = C2NDSource(self, max_concurrency=REMOTE_MAX_CONCURRENCY) - source.adopt_index(self._pending_index) + source._adopt_index(self._pending_index) return source except NotRanged as exc: # PartsMissing among them, which carries no status and so is not @@ -861,7 +1188,7 @@ def _open_block_source(self): # fields these read: whole chunks work for all of those return None - def adopt_index(self, state) -> None: + def _adopt_index(self, state) -> None: """Keep an index a `Proxy` read out of its cache until there is a source. Handing it straight to :meth:`block_source` would build the source to @@ -871,14 +1198,14 @@ def adopt_index(self, state) -> None: """ self._pending_index = state - def index_state(self, keep=()) -> dict | None: + def _index_state(self, keep=()) -> dict | None: """What a `Proxy` should keep of what was read; see :ref:`ByteRangeNDSource`.""" source = self._block_source if source is _UNTRIED or source is None: # No source was ever built, so nothing was read through one: hand back # whatever came out of the cache, rather than dropping it return self._pending_index - return source.index_state(keep) + return source._index_state(keep) def wants_blocks(self, nchunk: int, nwanted: int) -> bool: """Whether fetching *nwanted* blocks of a chunk beats fetching all of it.""" @@ -917,7 +1244,7 @@ def read_ranges(self, spans: Sequence[tuple[int, int]]) -> list[bytes]: return source.read_ranges(spans) @contextmanager - def _ranged(self): + def _ranged(self, index_only: bool = False): """The block source, retired if it turns out to serve ranges no longer. The subscriber can stop serving a dataset from a file between one fetch @@ -932,7 +1259,7 @@ def _ranged(self): chunks it was after whole, and a caller reading ranges directly is entitled to hear that the ranges are gone. """ - source = self.block_source() + source = self._index_source() if index_only else self.block_source() if source is None: # A `NotRanged`, which is a `ValueError`: a fetch that finds the # source retired under it -- by another thread of the same wave -- @@ -978,16 +1305,19 @@ def cparams(self) -> blosc2.CParams: @property def nbytes(self) -> int: """The number of bytes of the remote array""" + self._refresh_meta() return self.meta["schunk"]["nbytes"] @property def cbytes(self) -> int: """The number of compressed bytes of the remote array""" + self._refresh_meta() return self.meta["schunk"]["cbytes"] @property def cratio(self) -> float: """The compression ratio of the remote array""" + self._refresh_meta() return self.meta["schunk"]["cratio"] # TODO: Add these to SChunk model in srv_utils and then access them here @@ -1009,7 +1339,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 @@ -1054,6 +1390,7 @@ def info_items(self) -> list: @property def blocksize(self) -> int: """The block size (in bytes) for the remote container.""" + self._refresh_meta() return self.meta["schunk"]["blocksize"] diff --git a/src/blosc2/core.py b/src/blosc2/core.py index 6b36db274..856684d65 100644 --- a/src/blosc2/core.py +++ b/src/blosc2/core.py @@ -651,16 +651,17 @@ def normalize_urlpath(urlpath: object) -> object: def is_fsspec_url(urlpath: object) -> bool: """Whether *urlpath* should be routed through fsspec. - Any URL with a scheme qualifies, except `file://` (which the local path - handles better, with mmap and every container format) and `http(s)://` - (reserved for :ref:`C2Array`). Chained URLs such as + Any URL with a scheme qualifies, except `file://`, which the local path + handles better -- with mmap and every container format. Chained URLs such as `zip://x.b2nd::s3://bucket/a.zip` qualify too, as fsspec resolves them. + + `http(s)://` included: a frame behind a plain web server is a frame like any + other, and fsspec reads it in ranges wherever the server answers them. A + Caterva2 subscriber is not reached this way -- its datasets are named by root + and path rather than by URL, so :ref:`C2Array` is entered through + :ref:`URLPath`, which `open` dispatches on before it ever gets here. """ - return ( - isinstance(urlpath, str) - and "://" in urlpath - and not urlpath.startswith(("file://", "http://", "https://")) - ) + return isinstance(urlpath, str) and "://" in urlpath and not urlpath.startswith("file://") def _import_fsspec(urlpath: str): diff --git a/src/blosc2/proxy.py b/src/blosc2/proxy.py index 005539127..8aa897021 100644 --- a/src/blosc2/proxy.py +++ b/src/blosc2/proxy.py @@ -59,6 +59,14 @@ class Proxy(blosc2.Operand): :ref:`ProxySource` or :ref:`ProxyNDSource` interfaces. """ + _stamped = False + """Whether the source names the bytes it reads, as `_adopt_cache` found out. + + Kept because `_save_fetched` asks after every fetch and the answer cannot + change: a source either can name itself or cannot. Asking the source each + time would cost a request for one that has to look at its remote to answer. + """ + def __init__( self, src: ProxySource or ProxyNDSource, urlpath: str | None = None, mode="a", **kwargs: dict ): @@ -133,6 +141,15 @@ def __init__( f"for the proxy's own bookkeeping and cannot be set through vlmeta" ) + # Before either the cache is reopened or its stamp is judged: a source + # read once when it was opened names itself as it was then, and a handle + # that has outlived someone else's writes would hand over a stamp the + # cache still matches and a set of bytes it no longer does. Sources whose + # bytes cannot move underneath them do not offer this and are not asked + refresh = getattr(self.src, "refresh_stamp", None) + if refresh is not None: + refresh() + if self._cache is None and mode == "a" and urlpath is not None and os.path.exists(urlpath): # Reuse the cache left by an earlier run: whatever was fetched then is # still in there, and the creation path below would refuse to build @@ -236,6 +253,10 @@ def _adopt_cache(self, fresh: bool, nchunks: int) -> bytearray: `__getitem__`) rather than coming back stale. """ stamp = getattr(self.src, "stamp", None) + # Whether this source names itself at all, kept rather than asked again: + # reading the stamp of one that is being written to costs a request, and + # `_save_fetched` wants only the yes or no, after every fetch it makes + self._stamped = stamp is not None stored = None if fresh else self._schunk_cache.vlmeta.get("proxy-stamp") replaced = stamp is not None and stored is not None and stored != stamp writable = getattr(self._schunk_cache, "mode", None) != "r" @@ -257,7 +278,7 @@ def _adopt_cache(self, fresh: bool, nchunks: int) -> bytearray: # are, as an earlier run read them. Only from a cache that names the very # same remote bytes, checked here rather than taken on trust from how the # cache was come by: a `_cache=` handed in never passed `_reopen_cache`. - adopt = getattr(self.src, "adopt_index", None) + adopt = getattr(self.src, "_adopt_index", None) if adopt is not None and stamp is not None and stored == stamp: index = self._schunk_cache.vlmeta.get("proxy-index") adopt(index) @@ -419,8 +440,8 @@ def _save_fetched(self) -> None: # came from, and reusing them across a replacement is worse than serving # stale data. Bounded by keeping layouts for the partly filled chunks # alone, which are the only ones a later fetch would ask about. - state = getattr(self.src, "index_state", None) - if state is not None and getattr(self.src, "stamp", None) is not None: + state = getattr(self.src, "_index_state", None) + if state is not None and self._stamped: index = state(self._partly_filled()) # Only when it says something new: the offsets are the bulk of it and # never change once read, so a slice-by-slice walk would otherwise diff --git a/src/blosc2/proxy_source.py b/src/blosc2/proxy_source.py index 18648d3b0..a1f93dd03 100644 --- a/src/blosc2/proxy_source.py +++ b/src/blosc2/proxy_source.py @@ -333,6 +333,58 @@ async def aget_chunk(self, nchunk: int) -> bytes: _FRAME_MAGIC = b"b2frame\0" _CHUNK_HEADER_LEN = blosc2.MAX_OVERHEAD +# What a run-length offset codes in its top byte: the ones a frame writes are a +# run of zeros (1), of NaNs (2), and a chunk never written at all (4). The last +# is the only one that says "no content has ever been stored here", which is what +# `written_chunks` reads and what a pre-sized array is filled with +_SPECIAL_ZERO = 0x1 +_SPECIAL_NAN = 0x2 +_SPECIAL_UNINIT = 0x4 + + +def _special_kind(offset: int) -> int: + """Which run-length value a negative chunk offset codes.""" + return ((offset & 0xFFFFFFFFFFFFFFFF) >> 56) & 0x7 + + +def _special_kinds(offsets: np.ndarray) -> np.ndarray: + """The same for a whole index at once; see `_special_kind`. + + The offsets have to be in the host's own order for this: the tag lives in the + top byte of the word, and a view is what reads it, so an array that still + carries the byte order it was stored in would have the tag read out of the + wrong end. `_read_frame_offsets` and `_adopt_index` both hand over native + ones, which is what makes this the only place the two ever differ. + """ + return (offsets.view(np.uint64) >> np.uint64(56)) & np.uint64(0x7) + + +def _check_specials(offsets: np.ndarray, urlpath: str) -> None: + """Refuse a frame whose run-length offsets code something unknown. + + Here rather than in `_special_chunk`, which runs in the middle of a fetch: a + chunk this cannot rebuild is a property of the frame, and a fetch that meets + it half way through has no fallback for it -- `Proxy.fetch` gives way to + whole chunks for a `NotRanged`, and this is not one. Read once per index, + which is once per source unless it is written to. + """ + special = offsets < 0 + if not special.any(): + return + unknown = special & ~np.isin(_special_kinds(offsets), [_SPECIAL_ZERO, _SPECIAL_NAN, _SPECIAL_UNINIT]) + if unknown.any(): + nchunk = int(np.flatnonzero(unknown)[0]) + raise NotImplementedError( + f"chunk {nchunk} of {urlpath} has offset {int(offsets[nchunk])}, which codes " + f"run-length value {int(_special_kinds(offsets)[nchunk])}" + ) + + +def _section(layout: tuple) -> bytes: + """A chunk's header section, as the read that found its layout saw it.""" + head, bstarts, _ = layout + return head + bstarts.astype(" np.ndarray: """How many bytes each block of a chunk occupies, given where they start. @@ -561,6 +613,9 @@ def __init__(self, urlpath: str, max_concurrency: int = REMOTE_MAX_CONCURRENCY): # a b2nd metalayer -- is in the header that was just read. self._index = None self._index_lock = threading.Lock() + # Set when the frame is written to under this handle: the header moves as + # well as the offsets, so both are read again before the next lookup + self._stale = False try: _, _, shape, chunks, blocks, dtype_format, dtype = _frame_metalayer(raw, self._header, "b2nd") except KeyError: @@ -584,12 +639,12 @@ def __init__(self, urlpath: str, max_concurrency: int = REMOTE_MAX_CONCURRENCY): if all(self._blocks) else 1 ) - # Layouts are memoized for the life of the source; `index_state` hands + # Layouts are memoized for the life of the source; `_index_state` hands # them back as the bytes they were read as, so a `Proxy` can keep them in # its cache and a later run start from them instead of reading again. self._layouts = {} - def index_state(self, keep: Sequence[int] = ()) -> dict: + def _index_state(self, keep: Sequence[int] = ()) -> dict: """Where things are, as the bytes they were read as, for a cache to keep. The frame's chunk offsets, and the header sections of the chunks in @@ -606,27 +661,40 @@ def index_state(self, keep: Sequence[int] = ()) -> dict: second copy of every chunk ever laid out would grow for the life of the source to be read back a handful of chunks at a time. """ - offsets = self._index[0] if self._index is not None else None + with self._index_lock: + # Not while it is stale: what is kept here goes into a cache, and the + # next run adopts it against a stamp that says the array has not moved + # since -- which for a complete array is true of the array and false + # of these, so nothing would ever catch them. Handing back nothing + # costs that run a read of the offsets; handing back these would cost + # it a chunk that is in the frame and reads as never written + offsets = None if self._stale or self._index is None else self._index[0] return { "bpc": self.blocks_per_chunk, # Little-endian whatever the host is: a cache directory outlives the # machine that filled it, and a stamp cannot tell a byte order "offsets": b"" if offsets is None else offsets.astype(" bytes: - """A chunk's header section, as the read that found its layout saw it. + def _sections(self, keep: Sequence[int]): + """The layouts of *keep* that there are, paired with the chunk they are of. A chunk with no layout has none to give back, and none is wanted: a `Proxy` keeps layouts for the chunks it holds some blocks of, and a chunk that cannot be taken apart was fetched whole. """ - head, bstarts, _ = self._layouts[nchunk] - return head + bstarts.astype(" None: - """Take up what an earlier run left behind in :meth:`index_state`. + def _adopt_index(self, state: dict | None) -> None: + """Take up what an earlier run left behind in `_index_state`. Only ever called with a state saved against the very same remote bytes -- :ref:`Proxy` checks the source's ``stamp`` against the one its cache @@ -645,11 +713,22 @@ def adopt_index(self, state: dict | None) -> None: offsets = state.get("offsets") or b"" if offsets: nchunks = math.prod(math.ceil(s / c) for s, c in zip(self._shape, self._chunks, strict=True)) - array = np.frombuffer(offsets, dtype=" tuple[np.ndarray, np.ndarray]: no worse -- what they read is the same either way. """ 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, 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) + _check_specials(offsets, self.urlpath) self._index = (offsets, _chunk_extents(offsets, self._header)) self._head = None # the prefetch has nothing left to answer return self._index @@ -682,6 +771,57 @@ def _extents(self) -> np.ndarray: """How many bytes to read at each chunk's offset to be sure of covering it.""" return self._frame_index()[1] + def written_chunks(self) -> np.ndarray: + """Which chunks of the frame hold content, as a boolean per chunk. + + False only for a chunk that was never written: a frame keeps those in + their offset rather than in the file, tagged as uninitialized, which is + what `blosc2.uninit` fills an array with. Everything else is True, + a run of zeros included -- a writer that stored an all-zero chunk stored + something, and the tag says so, which is the whole reason to pre-size an + array with `uninit` rather than with `zeros`. + + One range read of the frame's offsets, and none at all once they have + been read: this is the same index every chunk read goes through. So the + progress of an array being filled is legible from the bytes a reader + already fetches, without asking the server anything about it. + """ + offsets = self._offsets + return ~((offsets < 0) & (_special_kinds(offsets) == _SPECIAL_UNINIT)) + + def invalidate_index(self) -> None: + """Forget where the chunks and blocks are, so the next read looks again. + + The frame's offsets move whenever it is written to: a chunk written into + a slot that held no content is appended past the old offsets block, which + the new one is then written after. Chunks already placed keep their + offsets -- that is what makes an append-only fill cheap to read + alongside -- but the index as a whole has to be read again to see the + slot that was filled, and the header with it, since the frame's length + and its payload extent are what the offsets are found through. + + Nothing is read here: the next lookup pays for it, so a writer that never + reads back spends no request on this at all. + + Only for a handle that writes, or that follows a frame someone else is + writing. A frame that nobody mutates never needs this. + """ + with self._index_lock: + # What was read stays until something reads again, so that a lookup + # racing this one is served the old positions rather than none at all; + # `_index_state` is what must not hand them on, and it asks about + # `_stale` for exactly that reason. + self._stale = True + # The layouts do go. Where a chunk is says nothing about whether the + # bytes at that position are still the ones its blocks were mapped + # from: an append-only fill leaves them alone, but a frame rewritten + # in place -- which this method's name promises nothing against -- + # keeps the offset and moves the block starts inside it, and a plan + # built from the old ones splices the wrong bytes into a chunk it + # then presents as whole. A layout costs one header read to rebuild + # and only the partly fetched chunks have one at all + self._layouts.clear() + @property def shape(self) -> tuple: return self._shape @@ -857,13 +997,15 @@ async def aget_chunk(self, nchunk: int) -> bytes: def _special_chunk(self, offset: int) -> bytes: """Rebuild a run-length chunk, which lives in its offset instead of the file.""" - kind = ((offset & 0xFFFFFFFFFFFFFFFF) >> 56) & 0x7 + kind = _special_kind(offset) nitems = self._chunksize // self._dtype.itemsize - if kind == 2: + if kind == _SPECIAL_NAN: data = np.full(nitems, np.nan, dtype=self._dtype) else: - # A run of zeros (1); uninitialized chunks (4) have no defined - # content, and zeros is what reading them locally hands back too + # A run of zeros; an uninitialized chunk has no defined content, and + # zeros is what reading one locally hands back too. Nothing else can + # arrive here -- `_check_specials` refuses the frame when the index is + # read, which is before any of this is asked for data = np.zeros(nitems, dtype=self._dtype) # 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 diff --git a/src/blosc2/schunk.py b/src/blosc2/schunk.py index c06ba2b9f..9f8b5d786 100644 --- a/src/blosc2/schunk.py +++ b/src/blosc2/schunk.py @@ -2085,9 +2085,11 @@ def open( ---------- urlpath: str | pathlib.Path | :ref:`URLPath` The path where the :ref:`SChunk` (or :ref:`NDArray`) - is stored. If it is a remote Caterva2 array, a :ref:`URLPath` must be passed. - Any other URL with a scheme (``s3://``, ``gs://``, ``zip://``, ``memory://``...) - is opened through fsspec; see the `Notes` section for the limits. + is stored. If it is a remote Caterva2 array, a :ref:`URLPath` must be passed: + a subscriber names its datasets by root and path rather than by URL. + Any URL with a scheme (``s3://``, ``gs://``, ``https://``, ``zip://``, + ``memory://``...) is opened through fsspec; see the `Notes` section for + the limits. mode: str, optional Persistence mode: 'r' means read only (must exist); 'a' means read/write (create if it doesn't exist); diff --git a/tests/ndarray/test_c2array_blocks.py b/tests/ndarray/test_c2array_blocks.py index e8a967e1b..0386dd4c8 100644 --- a/tests/ndarray/test_c2array_blocks.py +++ b/tests/ndarray/test_c2array_blocks.py @@ -408,9 +408,10 @@ def test_blocks_survive_a_reopened_cache(tmp_path, subscriber, any_chunk_wants_b def test_a_cache_that_holds_the_slice_costs_no_request(tmp_path, subscriber, any_chunk_wants_blocks): # Re-running a script over a cache that already covers the slice: `api/info` - # is all it takes. Nothing opens the frame, because opening it is what - # `block_source` puts off until a fetch actually wants a chunk -- and this - # fetch wants none. + # is all it takes -- the one the proxy spends looking again at an array that + # could have been written to since (see `refresh_stamp`). Nothing opens the + # frame, because opening it is what `block_source` puts off until a fetch + # actually wants a chunk -- and this fetch wants none. data = _incompressible((200, 200)) array, sub = subscriber(data, chunks=(100, 200), blocks=(10, 20)) cache = str(tmp_path / "held.b2nd") @@ -423,13 +424,13 @@ def test_a_cache_that_holds_the_slice_costs_no_request(tmp_path, subscriber, any p = blosc2.Proxy(again, urlpath=cache, mode="a") p.fetch(item) assert np.array_equal(p[item], data[item]) - assert not sub.log + assert [kind for kind, _, _ in sub.log] == ["info"] # ... and a slice the cache does not hold opens the frame then: the header, # the layout of the chunk it lands in, and the blocks. Not where the chunks # are -- the earlier run left that in the cache assert np.array_equal(p[100:105, 0:10], data[100:105, 0:10]) - assert [kind for kind, _, _ in sub.log] == ["fetch"] * 3 + assert [kind for kind, _, _ in sub.log] == ["info"] + ["fetch"] * 3 def test_a_kept_index_halves_a_warm_fetch(tmp_path, subscriber, any_chunk_wants_blocks): @@ -445,7 +446,8 @@ def test_a_kept_index_halves_a_warm_fetch(tmp_path, subscriber, any_chunk_wants_ sub.log.clear() p = blosc2.Proxy(again, urlpath=cache, mode="a") assert np.array_equal(p[:, 100:110], data[:, 100:110]) - assert [kind for kind, _, _ in sub.log] == ["fetch", "fetch"] # the header, the blocks + # The proxy's look at the array, then the header and the blocks -- nothing between + assert [kind for kind, _, _ in sub.log] == ["info", "fetch", "fetch"] assert np.array_equal(p[...], data) # ... and the rest still reads right @@ -463,9 +465,9 @@ def test_a_kept_index_does_not_open_the_frame_to_be_taken_up(tmp_path, subscribe sub.log.clear() p = blosc2.Proxy(again, urlpath=cache, mode="a") assert again._pending_index is not None # taken out of the cache, not yet used - assert not sub.log + assert [kind for kind, _, _ in sub.log] == ["info"] # the proxy's look, and no frame read p.fetch(item) - assert not sub.log + assert [kind for kind, _, _ in sub.log] == ["info"] def test_a_whole_chunk_cache_is_adopted(tmp_path, subscriber, any_chunk_wants_blocks): @@ -683,8 +685,9 @@ def test_a_cache_from_the_same_bytes_is_adopted(tmp_path, subscriber, any_chunk_ def test_no_stamp_when_the_subscriber_reports_no_mtime(tmp_path, subscriber): # Then the cache is checked on geometry alone, as every unstamped source is data = _incompressible((200, 200)) - array, _sub = subscriber(data, chunks=(100, 200), blocks=(10, 20)) - del array.meta["mtime"] + array, sub = subscriber(data, chunks=(100, 200), blocks=(10, 20)) + sub.mtime = None # the subscriber itself reports none, and goes on doing so + array = blosc2.C2Array(array.path, urlbase=array.urlbase) assert array.stamp is None cache = str(tmp_path / "unstamped.b2nd") diff --git a/tests/ndarray/test_c2array_writes.py b/tests/ndarray/test_c2array_writes.py new file mode 100644 index 000000000..8a9b1a3f1 --- /dev/null +++ b/tests/ndarray/test_c2array_writes.py @@ -0,0 +1,658 @@ +####################################################################### +# Copyright (c) 2019-present, Blosc Development Team +# All rights reserved. +# +# SPDX-License-Identifier: BSD-3-Clause +####################################################################### + +"""Filling a pre-sized remote array a chunk at a time, from several writers. + +The stand-in here answers the write contract a subscriber is meant to answer: +one chunk per request, into a slot nothing was written to yet, refused with a +409 otherwise. That refusal is the whole of the coordination -- the frame's own +offsets say which slots are free, so two writers that both believe they own a +chunk are resolved by the array rather than by anything either of them holds. +""" + +import concurrent.futures +import contextlib +import json +import os +import pathlib +import threading +import time +import uuid +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from urllib.parse import parse_qs, urlparse + +import numpy as np +import pytest + +import blosc2 + +# The stand-in binds a real socket, which wasm32 has no listen(2) for +pytestmark = pytest.mark.skipif(blosc2.IS_WASM, reason="no listening sockets on wasm32") + +CHUNKS = (1000,) +BLOCKS = (250,) +NCHUNKS = 6 +SHAPE = (CHUNKS[0] * NCHUNKS,) + + +class _Subscriber: + """A Caterva2-shaped server over one .b2nd file, that also accepts writes.""" + + def __init__(self, path): + self.path = str(path) + self.log = [] # (endpoint, status) + self.lock = threading.Lock() # what the real server does with holding_lock() + # One handle for the life of the server, and the only one in this process: + # a second handle open over a frame another is writing is the stale-handle + # hazard of `todo/locking-mwmr.md`, and it is silent -- the write reports + # nothing and the frame is left unreadable + self.array = blosc2.open(self.path, mode="a", locking=True) + self.reload() + + def reload(self): + self.mtime = pathlib.Path(self.path).stat().st_mtime + + @property + def meta(self): + array = self.array + schunk = array.schunk + return { + "shape": list(array.shape), + "chunks": list(array.chunks), + "blocks": list(array.blocks), + "dtype": str(array.dtype), + "mtime": self.mtime, + "schunk": { + "cparams": {"typesize": array.dtype.itemsize}, + "nbytes": schunk.nbytes, + "cbytes": schunk.cbytes, + "cratio": schunk.cratio, + "blocksize": schunk.blocksize, + "vlmeta": schunk.vlmeta.getall(), + }, + } + + def write_chunk(self, nchunk, chunk): + """The endpoint's body: refuse a slot that holds anything, then store. + + Serialized, as the server serializes it, and the whole of the check is + the slot's own tag: UNINIT and nothing else means never written, since a + writer that stored an all-zero chunk stored something. + """ + with self.lock: + array = self.array + infos = list(array.schunk.iterchunks_info()) + if not 0 <= nchunk < len(infos): + return 404, {"detail": "no such chunk"} + if infos[nchunk].special is not blosc2.SpecialValue.UNINIT: + return 409, {"detail": f"chunk {nchunk} was already written"} + nbytes = blosc2.get_cbuffer_sizes(chunk)[0] + if nbytes != array.schunk.chunksize: + return 400, {"detail": "the chunk does not match the array's chunkshape"} + array.schunk.update_chunk(nchunk, chunk) + vlmeta = array.schunk.vlmeta + if "fill_nonce" not in vlmeta.getall(): + # What names this array, as against another that comes to sit at + # the same path with the same size + vlmeta["fill_nonce"] = uuid.uuid4().hex + # Counted through the handle that wrote, rather than a fresh open of + # a frame the write just moved + written = sum( + 1 for i in array.schunk.iterchunks_info() if i.special is not blosc2.SpecialValue.UNINIT + ) + if written == len(infos) and vlmeta.getall().get("fill_state", "filling") == "filling": + vlmeta["fill_state"] = "complete" + self.reload() + return 200, {"written": written, "nchunks": len(infos), "nchunk": nchunk} + + +class _Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *args): + pass + + def handle(self): + with contextlib.suppress(ConnectionResetError, BrokenPipeError): + super().handle() + + def _send(self, status, body, headers=(), endpoint=""): + self.server.subscriber.log.append((endpoint, status)) + self.send_response(status) + for name, value in headers: + self.send_header(name, value) + self.send_header("Content-Length", str(len(body))) + self.end_headers() + self.wfile.write(body) + + def do_GET(self): + sub = self.server.subscriber + endpoint = self.path.split("/")[2] + if endpoint == "info": + self._send(200, json.dumps(sub.meta).encode(), endpoint="info") + elif endpoint == "chunk": + nchunk = int(self.path.split("nchunk=")[1]) + with sub.lock: + self._send(200, sub.array.schunk.get_chunk(nchunk), endpoint="chunk") + elif endpoint == "fetch": + self._fetch(sub) + else: + self._send(404, b"", endpoint=endpoint) + + def do_POST(self): + sub = self.server.subscriber + endpoint = self.path.split("/")[2].split("?")[0] + if endpoint != "chunk": + self._send(404, b"", endpoint=endpoint) + return + nchunk = int(self.path.split("nchunk=")[1]) + body = self.rfile.read(int(self.headers.get("Content-Length", 0))) + status, answer = sub.write_chunk(nchunk, body) + self._send(status, json.dumps(answer).encode(), endpoint="write") + + def _fetch(self, sub): + """Ranges over the frame's bytes, or the slice itself when none is asked. + + Both halves of what a subscriber serves: `C2Array.__getitem__` asks for a + slice and gets a cframe of it, while the block path asks for byte ranges + of the file. A fill has to be visible through both. + """ + query = parse_qs(urlparse(self.path).query) + frame = pathlib.Path(sub.path).read_bytes() + wanted = self.headers.get("Range") + if not wanted: + with sub.lock: + array = sub.array + sliced = array[_parse_slice(query.get("slice_", [""])[0], array.ndim)] + self._send(200, blosc2.asarray(sliced).to_cframe(), endpoint="fetch") + return + spans = [] + for span in wanted.removeprefix("bytes=").split(","): + first, _, last = span.partition("-") + spans.append((int(first), min(int(last), len(frame) - 1) if last else len(frame) - 1)) + # Sorted and merged, the way Starlette answers several ranges + spans.sort() + merged = [spans[0]] + for first, last in spans[1:]: + if first <= merged[-1][1] + 1: + merged[-1] = (merged[-1][0], max(merged[-1][1], last)) + else: + merged.append((first, last)) + if len(merged) == 1: + first, last = merged[0] + self._send( + 206, + frame[first : last + 1], + [ + ("Content-Range", f"bytes {first}-{last}/{len(frame)}"), + ("Accept-Ranges", "bytes"), + ], + endpoint="fetch", + ) + return + boundary = "c2boundary" + body = b"" + for first, last in merged: + body += ( + f"--{boundary}\r\nContent-Type: application/octet-stream\r\n" + f"Content-Range: bytes {first}-{last}/{len(frame)}\r\n\r\n" + ).encode() + body += frame[first : last + 1] + b"\r\n" + body += f"--{boundary}--\r\n".encode() + self._send( + 206, + body, + [("Content-Type", f"multipart/byteranges; boundary={boundary}"), ("Accept-Ranges", "bytes")], + endpoint="fetch", + ) + + +def _parse_slice(text, ndim): + """What `blosc2.slice_to_string` wrote, read back.""" + if not text: + return slice(None) + parts = [] + for part in text.split(","): + part = part.strip() + if ":" in part: + first, _, last = part.partition(":") + parts.append(slice(int(first) if first else None, int(last) if last else None)) + else: + parts.append(int(part)) + return tuple(parts) if len(parts) > 1 else parts[0] + + +@pytest.fixture +def subscriber(tmp_path): + """A pre-sized, unwritten array and a server over it.""" + path = tmp_path / "run.b2nd" + presized = blosc2.uninit(SHAPE, dtype=np.int32, chunks=CHUNKS, blocks=BLOCKS, urlpath=str(path)) + del presized # the server's handle is to be the only one over this file + sub = _Subscriber(path) + server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler) + server.subscriber = sub + threading.Thread(target=server.serve_forever, daemon=True).start() + urlbase = f"http://127.0.0.1:{server.server_address[1]}/" + try: + yield blosc2.C2Array("run.b2nd", urlbase=urlbase), sub + finally: + server.shutdown() + server.server_close() + + +def _chunk(nchunk, value=None): + """A chunk of the array's geometry, tagged by which chunk it is.""" + data = np.full(CHUNKS, nchunk if value is None else value, dtype=np.int32) + return blosc2.compress2(data, typesize=4, blocksize=BLOCKS[0] * 4) + + +def test_a_chunk_written_is_read_back(subscriber): + array, sub = subscriber + array.update_chunk(2, _chunk(2)) + assert np.all(array[2 * CHUNKS[0] : 3 * CHUNKS[0]] == 2) + # ... and nothing else was touched + assert np.all(array[0 : CHUNKS[0]] == 0) + + +def test_a_second_write_is_refused(subscriber): + array, sub = subscriber + array.update_chunk(1, _chunk(1)) + with pytest.raises(blosc2.ChunkAlreadyWritten): + array.update_chunk(1, _chunk(1, value=99)) + assert np.all(array[CHUNKS[0] : 2 * CHUNKS[0]] == 1) # the first write stands + + +def test_a_chunk_of_the_wrong_shape_is_refused(subscriber): + array, sub = subscriber + wrong = blosc2.compress2(np.zeros(CHUNKS[0] // 2, dtype=np.int32), typesize=4) + with pytest.raises(Exception): # noqa: B017 -- an HTTP 400, whatever httpx calls it + array.update_chunk(0, wrong) + assert not array.written_chunks().any() + + +def test_written_chunks_tracks_the_fill(subscriber): + array, sub = subscriber + assert list(array.written_chunks()) == [False] * NCHUNKS + array.update_chunk(3, _chunk(3)) + assert list(array.written_chunks()) == [False, False, False, True, False, False] + array.update_chunk(0, _chunk(0)) + assert list(array.written_chunks()) == [True, False, False, True, False, False] + + +def test_a_written_chunk_of_zeros_counts_as_written(subscriber): + """The reason a pre-sized array is filled with `uninit` and not with `zeros`. + + Compressing an all-zero buffer gives a run-length chunk, so a slot written + with one is special again -- but tagged as zeros, not as uninitialized, which + is what keeps it distinguishable from a slot nobody has reached yet. + """ + array, sub = subscriber + array.update_chunk(4, _chunk(4, value=0)) + assert array.written_chunks()[4] + assert np.all(array[4 * CHUNKS[0] : 5 * CHUNKS[0]] == 0) + with pytest.raises(blosc2.ChunkAlreadyWritten): + array.update_chunk(4, _chunk(4)) + + +def test_a_fill_leaves_the_chunks_before_it_where_they_were(subscriber): + """What makes an append-only fill cheap to read alongside.""" + array, sub = subscriber + array.update_chunk(0, _chunk(0, value=42)) + placed = array.get_chunk(0) + for nchunk in range(1, NCHUNKS): + array.update_chunk(nchunk, _chunk(nchunk)) + assert array.get_chunk(0) == placed + assert np.all(array[0 : CHUNKS[0]] == 42) + for nchunk in range(1, NCHUNKS): + assert np.all(array[nchunk * CHUNKS[0] : (nchunk + 1) * CHUNKS[0]] == nchunk) + + +def test_concurrent_writers_fill_the_array(subscriber): + array, sub = subscriber + urlbase = array.urlbase + + def fill(nchunk): + # A writer of its own, as a separate process would have + writer = blosc2.C2Array("run.b2nd", urlbase=urlbase) + writer.update_chunk(nchunk, _chunk(nchunk)) + return nchunk + + with concurrent.futures.ThreadPoolExecutor(max_workers=NCHUNKS) as pool: + assert sorted(pool.map(fill, range(NCHUNKS))) == list(range(NCHUNKS)) + + assert array.written_chunks().all() + expected = np.repeat(np.arange(NCHUNKS, dtype=np.int32), CHUNKS[0]) + np.testing.assert_array_equal(array[:], expected) + + +def test_two_writers_racing_for_one_chunk_leave_one_winner(subscriber): + array, sub = subscriber + urlbase = array.urlbase + barrier = threading.Barrier(2) + + def fill(value): + writer = blosc2.C2Array("run.b2nd", urlbase=urlbase) + barrier.wait() + try: + writer.update_chunk(5, _chunk(5, value=value)) + return "won" + except blosc2.ChunkAlreadyWritten: + return "lost" + + with concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: + outcomes = sorted(pool.map(fill, (7, 8))) + assert outcomes == ["lost", "won"] + stored = np.unique(array[5 * CHUNKS[0] : 6 * CHUNKS[0]]) + assert len(stored) == 1 + assert stored[0] in (7, 8) + + +def test_a_reader_sees_chunks_that_land_after_it_read(subscriber): + array, sub = subscriber + array.update_chunk(0, _chunk(0)) + assert np.all(array[0 : CHUNKS[0]] == 0) # reads, and indexes, the frame + array.update_chunk(1, _chunk(1)) + assert np.all(array[CHUNKS[0] : 2 * CHUNKS[0]] == 1) + + +@pytest.mark.asyncio +async def test_chunks_can_be_written_off_the_event_loop(subscriber): + array, sub = subscriber + answer = await array.aupdate_chunk(2, _chunk(2)) + assert answer["written"] == 1 + with pytest.raises(blosc2.ChunkAlreadyWritten): + await array.aupdate_chunk(2, _chunk(2)) + await array.aclose() + assert np.all(array[2 * CHUNKS[0] : 3 * CHUNKS[0]] == 2) + + +def _fill(array, values=None): + for nchunk in range(NCHUNKS): + array.update_chunk(nchunk, _chunk(nchunk, value=None if values is None else values)) + + +def test_a_filling_array_is_stamped_afresh_on_every_write(subscriber): + """A cache of an array still being filled has to be thrown away, not kept. + + What it holds of a chunk nobody had written is the zeros an unwritten chunk + reads as, and the run-length offset it had; once a writer fills that slot + both are wrong, and nothing in the cache tells them from the chunks that are + still good. + """ + array, sub = subscriber + stamps = [] + for nchunk in range(3): + array.update_chunk(nchunk, _chunk(nchunk)) + stamps.append(blosc2.C2Array("run.b2nd", urlbase=array.urlbase).stamp) + assert len(set(stamps)) == len(stamps) + + +def test_a_complete_array_keeps_one_stamp(subscriber): + """Once every slot is claimed the array cannot change, so a cache of it stands.""" + array, sub = subscriber + _fill(array) + + def stamp(): + return blosc2.C2Array("run.b2nd", urlbase=array.urlbase).stamp + + complete = stamp() + assert complete.startswith("n") + # An mtime that moved for reasons of its own is not a reason to refetch + os.utime(sub.path, (time.time() + 10, time.time() + 10)) + sub.reload() + assert stamp() == complete + + +def test_two_arrays_at_one_path_are_told_apart(subscriber, tmp_path): + """The hole a size and an mtime leave, which is what the nonce closes. + + Both arrays here are filled with constant chunks, so they compress to exactly + the same size; the mtime is then made equal by hand. Nothing but the nonce + separates them, and a cache of the first served against the second would be + wrong in every chunk. + """ + array, sub = subscriber + _fill(array, values=1) + first = blosc2.C2Array("run.b2nd", urlbase=array.urlbase) + first_stamp, first_size = first.stamp, pathlib.Path(sub.path).stat().st_size + + # A different array comes to sit at the same path, of the same size + replacement = tmp_path / "replacement.b2nd" + presized = blosc2.uninit(SHAPE, dtype=np.int32, chunks=CHUNKS, blocks=BLOCKS, urlpath=str(replacement)) + del presized + sub.array = blosc2.open(str(replacement), mode="a", locking=True) + sub.path = str(replacement) + for nchunk in range(NCHUNKS): + sub.write_chunk(nchunk, _chunk(nchunk, value=2)) + sub.reload() + + assert pathlib.Path(sub.path).stat().st_size == first_size # same bytes on disk + os.utime(sub.path, (first.meta["mtime"], first.meta["mtime"])) + sub.reload() + second = blosc2.C2Array("run.b2nd", urlbase=array.urlbase) + assert second.meta["mtime"] == first.meta["mtime"] # ... and the same mtime + assert second.stamp != first_stamp + + +def test_an_array_with_no_nonce_is_stamped_as_before(tmp_path): + """An ordinary dataset, never filled a chunk at a time, is unchanged by this.""" + path = tmp_path / "plain.b2nd" + blosc2.asarray(np.arange(4000, dtype=np.int32), chunks=(1000,), blocks=(250,), urlpath=str(path)) + sub = _Subscriber(path) + server = ThreadingHTTPServer(("127.0.0.1", 0), _Handler) + server.subscriber = sub + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + array = blosc2.C2Array("plain.b2nd", urlbase=f"http://127.0.0.1:{server.server_address[1]}/") + assert array.stamp == f"{sub.mtime}:{array.meta['schunk']['cbytes']}" + finally: + server.shutdown() + server.server_close() + + +def test_a_cache_of_a_complete_array_survives_a_second_run(subscriber, tmp_path): + """What the nonce is for: the finished array is the one read again and again. + + The cache is reopened after the array's mtime has moved under it, which is + what a republish or a copy does. Nothing was refetched -- the stamp says it + is the same array, and a complete one cannot have changed. + """ + array, sub = subscriber + _fill(array) + cache = str(tmp_path / "cache.b2nd") + proxy = blosc2.Proxy(blosc2.C2Array("run.b2nd", urlbase=array.urlbase), urlpath=cache, mode="w") + expected = proxy[:] + del proxy + + os.utime(sub.path, (time.time() + 10, time.time() + 10)) + sub.reload() + sub.log.clear() + 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 + + +@pytest.fixture +def blocks_are_worth_it(monkeypatch): + """Take the size threshold out of the way, so these small chunks use blocks. + + The block path is where the frame's index is read, and where a write that + moved it is either seen or not; the chunks here are a few KB, which blosc2 + would never split, so the threshold is what would keep the path untaken. + """ + monkeypatch.setattr(blosc2.proxy_source, "BLOCK_MIN_CBYTES", 0) + + +def test_blocks_of_a_chunk_written_since_the_index_was_read(subscriber, tmp_path, blocks_are_worth_it): + """A `Proxy` reading blocks has to see a slot that was filled under it. + + `__getitem__` asks the subscriber for a slice and never touches the frame, + so a read that goes through it says nothing about the index. This one goes + through the offsets, the chunk's block starts and a range read of the block. + """ + array, sub = subscriber + array.update_chunk(1, _chunk(1)) + proxy = blosc2.Proxy(array, urlpath=str(tmp_path / "blocks.b2nd"), mode="w") + assert array.serves_blocks + np.testing.assert_array_equal(proxy[CHUNKS[0] : CHUNKS[0] + 10], np.full(10, 1, dtype=np.int32)) + + array.update_chunk(2, _chunk(2)) + np.testing.assert_array_equal(proxy[2 * CHUNKS[0] : 2 * CHUNKS[0] + 10], np.full(10, 2, dtype=np.int32)) + + +def test_an_index_a_write_moved_is_not_handed_to_a_cache(subscriber, blocks_are_worth_it): + """What `_index_state` keeps is where the chunks are, which a write moves. + + A cache adopts these against a stamp that says the array has not changed + since -- and for a complete array that is true of the array and false of an + index read before the write that completed it. Nothing downstream can catch + that, so what is stale is not handed over at all. + """ + array, sub = subscriber + array.update_chunk(0, _chunk(0, value=7)) + array.chunk_layout(0) # builds the source and reads the frame's offsets + kept = array._index_state()["offsets"] + assert kept + + array.update_chunk(1, _chunk(1)) + assert not array._index_state()["offsets"] # they describe a frame that moved + assert array.written_chunks()[1] # read again ... + assert array._index_state()["offsets"] not in (b"", kept) # ... and worth keeping again + + +def test_a_cache_over_a_handle_that_outlived_a_write_is_not_kept(subscriber, tmp_path): + """A handle names the array as it last looked, and a proxy has to look again. + + `meta` is read when the handle is opened and never again of itself, so a + reader that has outlived someone else's chunks would hand a `Proxy` the stamp + of the array as it was -- which the cache built under that stamp matches, and + the bytes no longer do. + """ + array, sub = subscriber + reader = blosc2.C2Array("run.b2nd", urlbase=array.urlbase) + array.update_chunk(0, _chunk(0, value=4)) + cache = str(tmp_path / "outlived.b2nd") + proxy = blosc2.Proxy(reader, urlpath=cache, mode="w") + np.testing.assert_array_equal(proxy[0 : CHUNKS[0]], np.full(CHUNKS[0], 4, dtype=np.int32)) + del proxy + + array.update_chunk(1, _chunk(1)) # another writer, which this handle never hears of + with pytest.raises(ValueError, match="different remote bytes"): + blosc2.Proxy(reader, urlpath=cache, mode="a") + proxy = blosc2.Proxy(reader, urlpath=cache, mode="w") + np.testing.assert_array_equal(proxy[CHUNKS[0] : 2 * CHUNKS[0]], np.full(CHUNKS[0], 1, dtype=np.int32)) + + +def test_written_chunks_does_not_answer_out_of_a_proxy_cache(subscriber, tmp_path, blocks_are_worth_it): + """The one question whose whole point is what other writers have done. + + A `Proxy` hands its cached index to the array before there is a source to put + it in, and the source takes it up as it is built. A fill read through that + is the fill as of whenever the cache was written. + """ + array, sub = subscriber + array.update_chunk(0, _chunk(0)) + cache = str(tmp_path / "pending.b2nd") + blosc2.Proxy(array, urlpath=cache, mode="w")[0:10] # leaves the offsets in the cache + + reader = blosc2.C2Array("run.b2nd", urlbase=array.urlbase) + blosc2.Proxy(reader, urlpath=cache, mode="a") # takes them up, unread + assert reader._pending_index is not None + array.update_chunk(1, _chunk(1)) + assert list(reader.written_chunks()) == [True, True, False, False, False, False] + + +def test_a_writer_that_lost_a_race_stops_believing_what_it_read(subscriber): + """The refusal is the one answer that proves another writer moved the frame.""" + array, sub = subscriber + loser = blosc2.C2Array("run.b2nd", urlbase=array.urlbase) + before = loser.stamp + array.update_chunk(3, _chunk(3)) + with pytest.raises(blosc2.ChunkAlreadyWritten): + loser.update_chunk(3, _chunk(3, value=9)) + assert loser.stamp != before + assert loser.stamp == blosc2.C2Array("run.b2nd", urlbase=array.urlbase).stamp + + +def test_a_write_that_lands_while_the_handle_looks_is_not_forgotten(subscriber, monkeypatch): + """Reading `api/info` is a round trip, and a write of this handle's can land + inside it. Such an answer describes the array as it was before that write: + keeping it would leave the handle believing it is current with nothing left + to say otherwise. + """ + array, sub = subscriber + array.update_chunk(0, _chunk(0)) # the handle now has a look to catch up on + real, raced = blosc2.c2array.info, [] + + def racing_info(*args, **kwargs): + answer = real(*args, **kwargs) + if not raced: + raced.append(True) + array.update_chunk(1, _chunk(1)) # lands while the answer is on its way + return answer + + monkeypatch.setattr(blosc2.c2array, "info", racing_info) + array.stamp # noqa: B018 -- the look whose answer predates that write + assert raced + assert array.stamp == blosc2.C2Array("run.b2nd", urlbase=array.urlbase).stamp diff --git a/tests/test_fsspec.py b/tests/test_fsspec.py index b799d24c1..2edac6d5a 100644 --- a/tests/test_fsspec.py +++ b/tests/test_fsspec.py @@ -6,6 +6,9 @@ # LICENSE file in the root directory of this source tree) ####################################################################### +import contextlib +import functools +import http.server import os import pathlib import threading @@ -533,11 +536,63 @@ def test_unknown_protocol(): blosc2.open("nosuchproto://bucket/key.b2nd") -def test_http_does_not_reach_fsspec(): - # http(s) is reserved for Caterva2, which is entered through blosc2.URLPath; - # a bare URL keeps failing as a missing local path rather than being fetched - with pytest.raises(FileNotFoundError): - blosc2.open("http://localhost:1/foo.b2nd") +@pytest.mark.skipif(blosc2.IS_WASM, reason="no listening sockets on wasm32") +def test_http_url_is_read_through_fsspec(tmp_path): + # A frame behind a plain web server -- no Caterva2 there to ask anything of -- + # is a frame like any other: fsspec reads it in ranges wherever the server + # answers them, so a slice costs what it touches and not the whole file. + # A Caterva2 dataset is not reached this way; it needs a `blosc2.URLPath`. + pytest.importorskip("aiohttp") # what fsspec reads http(s) with + data = np.arange(40_000, dtype="i4").reshape(200, 200) + root = tmp_path / "www" + root.mkdir() + blosc2.asarray(data, chunks=(50, 200), blocks=(10, 100), urlpath=str(root / "big.b2nd")) + + with _ranged_server(root) as urlbase: + whole = blosc2.open(f"{urlbase}/big.b2nd") # fetched in one go, as s3:// is + assert np.array_equal(whole[:], data) + + lazy = blosc2.open(f"{urlbase}/big.b2nd", lazy=True, cache_storage=str(tmp_path / "cs")) + assert isinstance(lazy, blosc2.Proxy) + assert isinstance(lazy.src, blosc2.FsspecNDSource) + assert lazy.src.stamp is not None # so a cache of it can tell it has moved + assert np.array_equal(lazy[3:5, 100:120], data[3:5, 100:120]) + + +@contextlib.contextmanager +def _ranged_server(root): + """A web server over *root* that honours `Range`, which the stock one does not.""" + + class Ranged(http.server.SimpleHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, *args): + pass + + def do_GET(self): + span = self.headers.get("Range") + if not span: + return super().do_GET() + body = (root / self.path.lstrip("/")).read_bytes() + first, _, last = span.removeprefix("bytes=").partition("-") + first, last = int(first), int(last) if last else len(body) - 1 + part = body[first : last + 1] + self.send_response(206) + self.send_header("Content-Range", f"bytes {first}-{last}/{len(body)}") + self.send_header("Accept-Ranges", "bytes") + self.send_header("Content-Length", str(len(part))) + self.end_headers() + self.wfile.write(part) + return None + + handler = functools.partial(Ranged, directory=str(root)) + server = http.server.ThreadingHTTPServer(("127.0.0.1", 0), handler) + threading.Thread(target=server.serve_forever, daemon=True).start() + try: + yield f"http://127.0.0.1:{server.server_address[1]}" + finally: + server.shutdown() + server.server_close() def test_zip_store_needs_cache(tmp_path):