From dd90983835725a6e253d2520142224a1fc8927ab Mon Sep 17 00:00:00 2001 From: Peter Todd Date: Thu, 6 Aug 2026 10:29:36 +0100 Subject: [PATCH 1/2] feat(fizarrita): share in-flight chunk fetches and decodes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `getWorker` consulted the cache while building its task list and wrote back only after the worker returned, so nothing existed to join between "someone started fetching this chunk" and "the result is cacheable". Two overlapping calls — two viewports, a re-render arriving mid-flight — both missed, both fetched the same bytes, and both decoded them. A cache cannot close that window on its own: `ChunkCache` is synchronous and holds decoded chunks, so an entry appears only once a decode has finished. Adds a module-level map of in-flight chunk promises, keyed exactly like the cache, so concurrent readers of one chunk share one fetch and one decode. Two smaller changes fall out of the same window: - The cache is consulted again when a task starts, not only when the task list was built. A chunk another call finished in between is now picked up instead of being refetched and redecoded. This costs one extra `cache.get` per chunk that reaches the task stage, which is why the op counts in "custom cache implementation receives get/set calls" move from 4/6 to 6/8. - Only the producer writes to the cache. Letting every sharer re-`set` the same object would be a redundant write, and a cache with dispose semantics would see its own live entry displaced by itself. The SAB-without-cache path is untouched: it decodes straight into the calling read's SharedArrayBuffer using that read's mapping, so there is no standalone chunk to hand to anyone else. In-flight entries are dropped as soon as they settle, on both paths. Keeping a rejection would make one transient fetch failure permanent for that chunk; keeping a fulfilment would shadow the cache and pin chunks it had since evicted. A sharer does hold its worker slot while waiting, costing some parallelism — but that slot would otherwise have gone to a duplicate round-trip and a duplicate decompression of bytes already in flight, so no useful work is displaced. Co-Authored-By: Claude Opus 5 --- fizarrita/src/get-worker.ts | 196 ++++++++++++++++++---------- test/browser/zarrita-worker.spec.ts | 130 +++++++++++++++++- 2 files changed, 252 insertions(+), 74 deletions(-) diff --git a/fizarrita/src/get-worker.ts b/fizarrita/src/get-worker.ts index da9ca62..f5c341a 100644 --- a/fizarrita/src/get-worker.ts +++ b/fizarrita/src/get-worker.ts @@ -72,6 +72,52 @@ export function createCacheKey( return `${storeId}:${arr.path}:${chunkKey}` } +/** + * Chunk fetch+decode operations currently in flight, keyed exactly like the + * cache, so that concurrent readers of the same chunk share one of each. + * + * A cache alone cannot do this: `ChunkCache` is synchronous and holds decoded + * chunks, so nothing lands in it until a decode has already finished. Two + * `getWorker` calls that overlap — two viewports, a re-render arriving mid-flight + * — therefore both miss, both fetch, and both decode the very same bytes. This + * map is what closes the window between "someone started fetching this" and + * "the result is cacheable". + */ +const pendingChunks = new Map>>() + +/** + * Run `produce` once per key, handing concurrent callers the same promise. + * + * The entry is removed as soon as it settles, on both paths: keeping a rejection + * would make one transient fetch failure permanent for that chunk, and keeping a + * fulfilment would duplicate the cache while pinning chunks the cache has since + * evicted. Removal is conditional on the entry still being ours so a later + * attempt that already replaced it is not dropped by our own settlement. + */ +function shareInFlightChunk( + key: string, + produce: () => Promise>, +): Promise> { + const inFlight = pendingChunks.get(key) + if (inFlight) { + return inFlight as Promise> + } + + const promise = produce() + pendingChunks.set(key, promise as unknown as Promise>) + + const forget = () => { + if (pendingChunks.get(key) === (promise as unknown)) { + pendingChunks.delete(key) + } + } + // Both handlers swallow: this branch exists only to clean up, and the caller + // still receives `promise` itself and still sees the rejection. + promise.then(forget, forget) + + return promise +} + // --------------------------------------------------------------------------- // Unified metadata reader — reads zarr.json once, returns everything needed // --------------------------------------------------------------------------- @@ -770,79 +816,84 @@ export async function getWorker< continue } + /** The zero/fill-value chunk used when the store has no bytes for this key. */ + const buildFillChunk = (): Chunk => { + const fillChunkShape = edgeChunkShape + const fillChunkStrides = get_strides(fillChunkShape) + const fillChunkSize = fillChunkShape.reduce( + (a: number, b: number) => a * b, + 1, + ) + const chunkData = new Ctr(fillChunkSize) + if (fillValue != null) { + // @ts-expect-error: fill_value type is union + chunkData.fill(fillValue) + } + return { + data: chunkData as Chunk["data"], + shape: fillChunkShape, + stride: fillChunkStrides, + } + } + tasks.push(async (workerSlot: WorkerLike | null) => { const worker = workerSlot ?? createCodecWorker(workerUrl) - // Fetch raw bytes from store on main thread - const rawBytes = await arr.store.get(chunkPath, opts.opts) - - if (!rawBytes) { - // Missing chunk — fill value, no worker needed - const fillChunkShape = edgeChunkShape - const fillChunkStrides = get_strides(fillChunkShape) - const fillChunkSize = fillChunkShape.reduce( - (a: number, b: number) => a * b, - 1, - ) - const chunkData = new Ctr(fillChunkSize) - if (fillValue != null) { - // @ts-expect-error: fill_value type is union - chunkData.fill(fillValue) - } - const chunk: Chunk = { - data: chunkData as Chunk["data"], - shape: fillChunkShape, - stride: fillChunkStrides, - } - // Cache the fill-value chunk - cache.set(cacheKey, chunk) - // Copy fill-value chunk into output on main thread - setter.set_from_chunk(out, chunk, mapping) - } else if (useShared && !opts.cache) { - // SAB path (no cache): worker decodes AND writes directly into the - // SharedArrayBuffer output — no transfer back, no main-thread copy. - try { - await workerDecodeInto( - worker, - rawBytes, - metaId, - correctedCodecMeta, - buffer as SharedArrayBuffer, - size * bytesPerElement, - outStride, - mapping, - bytesPerElement, - isEdgeChunk ? edgeChunkShape : undefined, - ) - } catch (error) { - worker.terminate() - throw error + // SAB path (no cache): the worker decodes AND writes directly into *this* + // call's SharedArrayBuffer, using *this* call's mapping — no transfer + // back, no main-thread copy. There is no standalone chunk here to hand to + // anyone else, so this path cannot participate in sharing and is left + // exactly as it was. + if (useShared && !opts.cache) { + const rawBytes = await arr.store.get(chunkPath, opts.opts) + if (!rawBytes) { + setter.set_from_chunk(out, buildFillChunk(), mapping) + } else { + try { + await workerDecodeInto( + worker, + rawBytes, + metaId, + correctedCodecMeta, + buffer as SharedArrayBuffer, + size * bytesPerElement, + outStride, + mapping, + bytesPerElement, + isEdgeChunk ? edgeChunkShape : undefined, + ) + } catch (error) { + worker.terminate() + throw error + } } - } else if (useShared && opts.cache) { - // SAB path with cache: use workerDecode to get a standalone chunk - // so we can cache it, then copy into the SAB output. The small - // overhead of transfer + copy on first access is repaid by - // subsequent cache hits that skip the worker entirely. - let chunk: Chunk - try { - chunk = await workerDecode( - worker, - rawBytes, - metaId, - correctedCodecMeta, - isEdgeChunk ? edgeChunkShape : undefined, - ) - } catch (error) { - worker.terminate() - throw error + return { worker, result: undefined as void } + } + + // The cache is consulted again here, not just when the task list was + // built: between those two moments another task — very likely a + // concurrent `getWorker` — may have finished this exact chunk. + const cachedSinceBuild = cache.get(cacheKey) + if (cachedSinceBuild) { + setter.set_from_chunk(out, cachedSinceBuild as Chunk, mapping) + return { worker, result: undefined as void } + } + + // One fetch and one decode per chunk, however many callers want it. The + // follower still holds its worker slot while waiting, which costs some + // parallelism — but that slot would otherwise have been spent on a + // duplicate network round-trip and a duplicate decompression of bytes + // already in flight, so it is not work being given up. + const chunk = await shareInFlightChunk(cacheKey, async () => { + const rawBytes = await arr.store.get(chunkPath, opts.opts) + if (!rawBytes) { + const fillChunk = buildFillChunk() + cache.set(cacheKey, fillChunk) + return fillChunk } - cache.set(cacheKey, chunk) - setter.set_from_chunk(out, chunk, mapping) - } else { - // Standard path: worker decodes, transfers back, main thread copies - let chunk: Chunk + let decoded: Chunk try { - chunk = await workerDecode( + decoded = await workerDecode( worker, rawBytes, metaId, @@ -853,9 +904,14 @@ export async function getWorker< worker.terminate() throw error } - cache.set(cacheKey, chunk) - setter.set_from_chunk(out, chunk, mapping) - } + // Cached by the producer only. Letting every sharer re-`set` the same + // object would be a redundant write, and a cache with dispose semantics + // would see its own live entry displaced by itself. + cache.set(cacheKey, decoded) + return decoded + }) + + setter.set_from_chunk(out, chunk, mapping) return { worker, result: undefined as void } }) diff --git a/test/browser/zarrita-worker.spec.ts b/test/browser/zarrita-worker.spec.ts index 3bb6001..41f1be4 100644 --- a/test/browser/zarrita-worker.spec.ts +++ b/test/browser/zarrita-worker.spec.ts @@ -1914,10 +1914,18 @@ test.describe('@fideus-labs/fizarrita — getWorker / setWorker', () => { expect(result.hasGetOps).toBe(true) expect(result.hasSetOps).toBe(true) - // First call: 2 get misses + 2 set calls = 4 ops - expect(result.opsAfterFirst.length).toBe(4) - // Second call: 2 get hits (no sets since already cached) - expect(result.opsAll.length).toBe(6) + // First call, per chunk: one get when the task list is built (miss), a + // second get when the task actually runs (still a miss here — nothing else + // is in flight), then one set. Two chunks, so 6 ops. + // + // The second get is what lets a task pick up a chunk that a concurrent + // `getWorker` finished in the interval between the list being built and the + // task starting — without it, that task refetches and redecodes something + // already sitting in the cache. + expect(result.opsAfterFirst.length).toBe(6) + // Second call: 2 get hits at build time, so no task is created and no + // second get or set happens. + expect(result.opsAll.length).toBe(8) }) test('cache with sliced access shares cached chunks', async ({ page }) => { @@ -2224,4 +2232,118 @@ test.describe('@fideus-labs/fizarrita — getWorker / setWorker', () => { expect(result.cacheSizeAfterSecond).toBe(2) }) + + test('concurrent reads of the same chunk fetch and decode it once', async ({ page }) => { + const result = await page.evaluate(async () => { + const { zarr, getWorker, WorkerPool } = window + // Wide enough that every task of every caller gets a slot at once — + // otherwise the first caller's tasks occupy the pool, settle, and the + // later callers never overlap with them at all. + const pool = new WorkerPool(8) + + const store = zarr.root() + const arr = await zarr.create(store, { + shape: [4], + chunk_shape: [2], + data_type: 'int32', + }) + await zarr.set(arr, null, { + data: new Int32Array([10, 20, 30, 40]), + shape: [4], + stride: [1], + }) + + // Hold reads of c/1 open until every caller has asked for it. Only c/1 is + // gated: the shape probe reads c/0 before any task runs, so gating that + // would stall all three callers before they reached the task phase and + // nothing would ever overlap. + const originalGet = arr.store.get.bind(arr.store) + const chunkPaths: string[] = [] + let release: () => void = () => {} + const gate = new Promise((resolve) => { + release = resolve + }) + ;(arr.store as any).get = async (path: string, ...rest: any[]) => { + if (path.includes('/c/')) { + chunkPaths.push(path) + if (path.endsWith('/c/1')) await gate + } + return originalGet(path, ...rest) + } + + const reads = [ + getWorker(arr, null, { pool }), + getWorker(arr, null, { pool }), + getWorker(arr, null, { pool }), + ] + // Let all three pile up on the gate, then let them through together. + await new Promise((resolve) => setTimeout(resolve, 100)) + release() + const chunks = await Promise.all(reads) + + pool.terminateWorkers() + + return { + // Sharing one decode must not mean sharing a half-filled output. + values: chunks.map((chunk) => Array.from(chunk.data as Int32Array)), + c1Reads: chunkPaths.filter((path) => path.endsWith('/c/1')).length, + } + }) + + const expected = [10, 20, 30, 40] + expect(result.values).toEqual([expected, expected, expected]) + // Three concurrent callers, one fetch. Without dedup this is 3. + expect(result.c1Reads).toBe(1) + }) + + test('a failed chunk fetch is not remembered by the in-flight map', async ({ page }) => { + const result = await page.evaluate(async () => { + const { zarr, getWorker, WorkerPool } = window + const pool = new WorkerPool(2) + + const store = zarr.root() + const arr = await zarr.create(store, { + shape: [4], + chunk_shape: [2], + data_type: 'int32', + }) + await zarr.set(arr, null, { + data: new Int32Array([10, 20, 30, 40]), + shape: [4], + stride: [1], + }) + + // Fail c/1 once. Not c/0: the shape probe reads that first and swallows + // its errors, so the failure would never reach a chunk task. + const originalGet = arr.store.get.bind(arr.store) + let failNext = true + ;(arr.store as any).get = async (path: string, ...rest: any[]) => { + if (path.endsWith('/c/1') && failNext) { + failNext = false + throw new Error('transient store failure') + } + return originalGet(path, ...rest) + } + + let firstFailed = false + try { + await getWorker(arr, null, { pool }) + } catch { + firstFailed = true + } + + // The store works again. A retry must not replay a held-onto rejection. + const retry = await getWorker(arr, null, { pool }) + pool.terminateWorkers() + + return { + firstFailed, + retryValues: Array.from(retry.data as Int32Array), + } + }) + + expect(result.firstFailed).toBe(true) + expect(result.retryValues).toEqual([10, 20, 30, 40]) + }) + }) From ebe8bd114b7ad4555c43a8f4d9f5c98b4e6ddfee Mon Sep 17 00:00:00 2001 From: Peter Todd Date: Fri, 7 Aug 2026 11:34:12 +0100 Subject: [PATCH 2/2] fix(fizarrita): populate every caller's cache, not just the producer's MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review catch: sharing was keyed on the chunk while the `cache.set` lived inside the producer, so a read that shared someone else's in-flight chunk never had its own cache filled. Worst shape is a cache-holding read sharing with a no-cache one — it asked for caching and silently got none for that chunk, which is the documented `cache` contract ("on a cache miss the decoded chunk is stored for future use") quietly not holding. Each caller now writes its own cache once the chunk arrives. The write is guarded by a lookup rather than unconditional: with a shared chunk object, a second write does not merely repeat itself, it displaces a live entry with itself, which a cache that disposes on overwrite would act on. Keeps the pending map keyed on the chunk rather than scoping it per cache. The expensive half is the store round-trip and the decode, and neither belongs to a particular cache — scoping the key would make two readers holding different caches fetch and decode the same bytes twice to arrive at the same chunk. Adds "every concurrent caller gets its own cache populated" — two overlapping reads with different caches, asserting one fetch, both caches filled, and the same chunk object in each. Verified to fail before this commit. The op counts in "custom cache implementation receives get/set calls" move again, to 8/10, for the third lookup this adds. Also replaces the fixed 100 ms sleep in the concurrency test with condition-based waits. The gated-request count cannot be the signal — dedup working means only one request reaches the gate — so it waits on the shape probe instead, which runs once per call outside the task path and so survives dedup. Co-Authored-By: Claude Opus 5 --- fizarrita/src/get-worker.ts | 28 ++++-- test/browser/zarrita-worker.spec.ts | 140 +++++++++++++++++++++++++--- 2 files changed, 143 insertions(+), 25 deletions(-) diff --git a/fizarrita/src/get-worker.ts b/fizarrita/src/get-worker.ts index f5c341a..8a29416 100644 --- a/fizarrita/src/get-worker.ts +++ b/fizarrita/src/get-worker.ts @@ -887,13 +887,10 @@ export async function getWorker< const chunk = await shareInFlightChunk(cacheKey, async () => { const rawBytes = await arr.store.get(chunkPath, opts.opts) if (!rawBytes) { - const fillChunk = buildFillChunk() - cache.set(cacheKey, fillChunk) - return fillChunk + return buildFillChunk() } - let decoded: Chunk try { - decoded = await workerDecode( + return await workerDecode( worker, rawBytes, metaId, @@ -904,13 +901,24 @@ export async function getWorker< worker.terminate() throw error } - // Cached by the producer only. Letting every sharer re-`set` the same - // object would be a redundant write, and a cache with dispose semantics - // would see its own live entry displaced by itself. - cache.set(cacheKey, decoded) - return decoded }) + // Populate *this* read's cache, whoever produced the chunk. Sharing is + // keyed on the chunk, not on the cache, so the producer may have been a + // concurrent read holding a different cache instance — or none at all. + // Leaving the write to the producer would mean a caller that supplied a + // cache silently not getting it filled, which is the `cache` contract + // ("on a cache miss the decoded chunk is stored for future use") quietly + // not holding. + // + // Guarded rather than unconditional so no cache is handed an entry it + // already holds: with a shared chunk that write is not merely redundant, + // it displaces a live entry with itself, which a cache that disposes on + // overwrite would act on. + if (!cache.get(cacheKey)) { + cache.set(cacheKey, chunk) + } + setter.set_from_chunk(out, chunk, mapping) return { worker, result: undefined as void } diff --git a/test/browser/zarrita-worker.spec.ts b/test/browser/zarrita-worker.spec.ts index 41f1be4..5b2d174 100644 --- a/test/browser/zarrita-worker.spec.ts +++ b/test/browser/zarrita-worker.spec.ts @@ -1914,18 +1914,19 @@ test.describe('@fideus-labs/fizarrita — getWorker / setWorker', () => { expect(result.hasGetOps).toBe(true) expect(result.hasSetOps).toBe(true) - // First call, per chunk: one get when the task list is built (miss), a - // second get when the task actually runs (still a miss here — nothing else - // is in flight), then one set. Two chunks, so 6 ops. - // - // The second get is what lets a task pick up a chunk that a concurrent - // `getWorker` finished in the interval between the list being built and the - // task starting — without it, that task refetches and redecodes something - // already sitting in the cache. - expect(result.opsAfterFirst.length).toBe(6) - // Second call: 2 get hits at build time, so no task is created and no - // second get or set happens. - expect(result.opsAll.length).toBe(8) + // First call does three gets and a set per chunk, each lookup avoiding + // strictly more expensive work than itself: + // 1. building the task list — a hit means no task at all; + // 2. starting the task — a hit means no fetch and no decode, catching a + // chunk a concurrent `getWorker` finished in between; + // 3. after the chunk arrives — decides whether this cache still needs it, + // since the chunk may have been produced for a different cache, or for + // none, and must not be written twice. + // All three miss here (nothing else is running), so 4 ops x 2 chunks = 8. + expect(result.opsAfterFirst.length).toBe(8) + // Second call: 2 get hits at build time, so no task is created and neither + // the later gets nor the sets happen. + expect(result.opsAll.length).toBe(10) }) test('cache with sliced access shares cached chunks', async ({ page }) => { @@ -2259,6 +2260,7 @@ test.describe('@fideus-labs/fizarrita — getWorker / setWorker', () => { // nothing would ever overlap. const originalGet = arr.store.get.bind(arr.store) const chunkPaths: string[] = [] + let gatedRequests = 0 let release: () => void = () => {} const gate = new Promise((resolve) => { release = resolve @@ -2266,18 +2268,43 @@ test.describe('@fideus-labs/fizarrita — getWorker / setWorker', () => { ;(arr.store as any).get = async (path: string, ...rest: any[]) => { if (path.includes('/c/')) { chunkPaths.push(path) - if (path.endsWith('/c/1')) await gate + if (path.endsWith('/c/1')) { + gatedRequests += 1 + await gate + } } return originalGet(path, ...rest) } + const waitFor = async (ready: () => boolean, label: string) => { + const deadline = Date.now() + 5000 + while (!ready()) { + if (Date.now() > deadline) throw new Error(`timed out waiting for ${label}`) + await new Promise((resolve) => setTimeout(resolve, 5)) + } + } + const reads = [ getWorker(arr, null, { pool }), getWorker(arr, null, { pool }), getWorker(arr, null, { pool }), ] - // Let all three pile up on the gate, then let them through together. - await new Promise((resolve) => setTimeout(resolve, 100)) + + // Release only once every caller is demonstrably at the point of wanting + // c/1, rather than after a fixed sleep. + // + // The gated-request count cannot be the signal — dedup working means only + // one request ever reaches the gate, so waiting for three would hang on a + // passing run. The shape probe is the observable that survives dedup: it + // reads c/0 once per call, outside the task path, before that call builds + // any tasks. Three c/0 reads therefore means all three callers are past + // probing and into their task phase, and a task's first act is the + // store.get for its chunk — no I/O in between. + await waitFor( + () => chunkPaths.filter((path) => path.endsWith('/c/0')).length >= 3, + 'all three callers to finish probing', + ) + await waitFor(() => gatedRequests >= 1, 'the shared c/1 fetch to begin') release() const chunks = await Promise.all(reads) @@ -2346,4 +2373,87 @@ test.describe('@fideus-labs/fizarrita — getWorker / setWorker', () => { expect(result.retryValues).toEqual([10, 20, 30, 40]) }) + + test('every concurrent caller gets its own cache populated', async ({ page }) => { + const result = await page.evaluate(async () => { + const { zarr, getWorker, WorkerPool } = window + const pool = new WorkerPool(8) + + const store = zarr.root() + const arr = await zarr.create(store, { + shape: [4], + chunk_shape: [2], + data_type: 'int32', + }) + await zarr.set(arr, null, { + data: new Int32Array([10, 20, 30, 40]), + shape: [4], + stride: [1], + }) + + const originalGet = arr.store.get.bind(arr.store) + const chunkPaths: string[] = [] + let gatedRequests = 0 + let release: () => void = () => {} + const gate = new Promise((resolve) => { + release = resolve + }) + ;(arr.store as any).get = async (path: string, ...rest: any[]) => { + if (path.includes('/c/')) { + chunkPaths.push(path) + if (path.endsWith('/c/1')) { + gatedRequests += 1 + await gate + } + } + return originalGet(path, ...rest) + } + + const waitFor = async (ready: () => boolean, label: string) => { + const deadline = Date.now() + 5000 + while (!ready()) { + if (Date.now() > deadline) throw new Error(`timed out waiting for ${label}`) + await new Promise((resolve) => setTimeout(resolve, 5)) + } + } + + // Two readers of the same chunk holding *different* caches. Sharing is + // keyed on the chunk, so only one of them fetches and decodes — but both + // asked for caching, so both caches have to end up holding the result. + const cacheA = new Map() + const cacheB = new Map() + const reads = [ + getWorker(arr, null, { pool, cache: cacheA }), + getWorker(arr, null, { pool, cache: cacheB }), + ] + + await waitFor( + () => chunkPaths.filter((path) => path.endsWith('/c/0')).length >= 2, + 'both callers to finish probing', + ) + await waitFor(() => gatedRequests >= 1, 'the shared c/1 fetch to begin') + release() + await Promise.all(reads) + + pool.terminateWorkers() + + const c1Key = [...cacheA.keys()].find((key) => key.endsWith('c/1')) + + return { + c1Reads: chunkPaths.filter((path) => path.endsWith('/c/1')).length, + cacheASize: cacheA.size, + cacheBSize: cacheB.size, + // The very same decoded chunk, not two copies of it. + sameChunkObject: !!c1Key && cacheA.get(c1Key) === cacheB.get(c1Key), + } + }) + + // Still deduped: one fetch across both callers. + expect(result.c1Reads).toBe(1) + // ...and neither caller was silently denied its cache entry. + expect(result.cacheASize).toBe(2) + expect(result.cacheBSize).toBe(2) + expect(result.sameChunkObject).toBe(true) + }) + })