From c34961fc13185a69883de2960da10da8e4cf204c Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 23 Oct 2026 18:01:49 +0000 Subject: [PATCH 1/4] Lightsheet: build each level as one dask layer; log where tile time goes Each pyramid level was one from_delayed array per chunk joined with da.block, so the graph had one layer per chunk and dask walked all of them on every slice. Showing 16 cached chunks cost ~1.4 s on a 16k-chunk level and ~26 s on a 131k-chunk level, against ~15 ms to read the files. A level is now a single map_blocks layer; the same view takes ~20 ms at any level size. Also print one [lightsheet] tiles: line per burst of reads, splitting tile time into path lookup (memo vs fetcher queue), disk read, decode and the burst span. NDI_LIGHTSHEET_TIMING=0 turns it off. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EJ9PUeM7x2vLsg8FcZQfXo --- CHANGELOG.md | 12 ++ src/ndi/pyramid/multiscale.py | 268 ++++++++++++++++++++++----- src/ndi/pyramid/upsample_fallback.py | 2 +- 3 files changed, 230 insertions(+), 52 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c9c3e892..dbe7447f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -104,6 +104,18 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). ### Fixed +- **Lightsheet viewer: slow tiles from the local cache.** Each pyramid level was + built as one dask graph layer per chunk (`from_delayed` + `da.block`), and + dask walks every layer each time napari computes a slice, so the cost grew + with the size of the level rather than the view: 16 cached chunks took ~1.4 s + to show on a 16k-chunk level and ~26 s on a 131k-chunk one, against ~15 ms to + read the files. This is the "Loading view 50/112" with nothing in flight. + A level is now a single `map_blocks` layer: the same view takes ~20 ms at + any level size, and building the level at open time is ~20x faster. + The viewer also now prints one `[lightsheet] tiles:` line per burst of reads + showing where tile time goes (path lookup, queue wait, disk read, decode, + and the burst's wall-clock span); `NDI_LIGHTSHEET_TIMING=0` turns it off. + - **Saving preferences no longer drops NDI-matlab's own entries.** `~/.ndi/NDI_Preferences.json` is shared with NDI-matlab, and each side rewrote it with only the preferences it registers, so saving from Python diff --git a/src/ndi/pyramid/multiscale.py b/src/ndi/pyramid/multiscale.py index f3fd1af2..55055a16 100644 --- a/src/ndi/pyramid/multiscale.py +++ b/src/ndi/pyramid/multiscale.py @@ -80,6 +80,142 @@ def __init__(self, key, doc, filename): self.callbacks: list = [] +class _TileTiming: + """Where each tile's time goes, printed as one line per burst of reads. + + Every chunk napari reads passes through :func:`_read_chunk_from_fetcher`, + which reports four phases here: finding the chunk's local path (a memo + hit, or a trip through the fetcher queue -- which also covers a disk + lookup or a cloud download), reading the bytes, decoding them, and the + whole tile. Once a burst of reads goes quiet, or every few seconds + during a long one, one ``[lightsheet] tiles:`` line goes to stderr. + + ``span`` is the wall time from the first tile starting to the last one + finishing. If span is much larger than the per-tile times can explain, + the time is going somewhere other than our reader (dask scheduling, + napari). ``NDI_LIGHTSHEET_TIMING=0`` turns the lines off. + """ + + _QUIET_S = 0.5 # a burst ends after this long with no tile finishing + _MAX_S = 5.0 # a long burst still reports this often + + def __init__(self, fetcher): + self._fetcher = fetcher + self._lock = threading.Lock() + self._on = os.environ.get("NDI_LIGHTSHEET_TIMING", "1").strip().lower() not in ( + "0", + "false", + "off", + ) + self._reset() + self._timer = None + + def _reset(self): + self.n = 0 + self.memo = 0 + self.queued = 0 + self.queued_s = 0.0 + self.queued_max = 0.0 + self.missing = 0 + self.read_s = 0.0 + self.read_bytes = 0 + self.decode_s = 0.0 + self.tile_s = 0.0 + self.tile_max = 0.0 + self.first = None + self.last = None + self.worker_base = self._workerCounts() + + def _workerCounts(self): + f = self._fetcher + return ( + getattr(f, "_direct_hits", 0), + getattr(f, "_direct_fetches", 0), + getattr(f, "_fallbacks", 0), + getattr(f, "_stale_released", 0), + ) + + def record(self, t0, memo_hit, path_s, found, read_s, nbytes, decode_s): + if not self._on: + return + now = time.monotonic() + with self._lock: + if self.first is None: + self.first = t0 + self.last = now + self.n += 1 + if memo_hit: + self.memo += 1 + else: + self.queued += 1 + self.queued_s += path_s + self.queued_max = max(self.queued_max, path_s) + if not found: + self.missing += 1 + self.read_s += read_s + self.read_bytes += nbytes + self.decode_s += decode_s + tile = now - t0 + self.tile_s += tile + self.tile_max = max(self.tile_max, tile) + if now - self.first >= self._MAX_S: + line = self._lineLocked() + else: + line = None + self._armLocked() + if line: + _printTiming(line) + + def _armLocked(self): + # One quiet-period timer at a time; it re-checks on firing. + if self._timer is None: + self._timer = threading.Timer(self._QUIET_S, self._quiet) + self._timer.daemon = True + self._timer.start() + + def _quiet(self): + with self._lock: + self._timer = None + if self.n == 0: + return + if time.monotonic() - self.last < self._QUIET_S: + self._armLocked() + return + line = self._lineLocked() + _printTiming(line) + + def _lineLocked(self) -> str: + n = self.n + hits, fetches, fallbacks, stale = ( + a - b for a, b in zip(self._workerCounts(), self.worker_base) + ) + span = self.last - self.first + mb = self.read_bytes / 1e6 + rate = f", {mb / self.read_s:.0f} MB/s" if self.read_s > 0 else "" + q_mean = 1000.0 * self.queued_s / self.queued if self.queued else 0.0 + line = ( + f"tiles: {n} in {span:.2f}s span | " + f"path: {self.memo} memo, {self.queued} via queue " + f"(mean {q_mean:.1f}ms, max {1000.0 * self.queued_max:.0f}ms; " + f"workers did {hits} disk hits, {fetches} cloud fetches, " + f"{fallbacks} via session, {stale} stale) | " + f"read {1000.0 * self.read_s / n:.1f}ms/tile ({mb:.1f} MB{rate}) | " + f"decode {1000.0 * self.decode_s / n:.1f}ms/tile | " + f"tile mean {1000.0 * self.tile_s / n:.1f}ms, max {1000.0 * self.tile_max:.0f}ms" + + (f" | {self.missing} not found" if self.missing else "") + ) + self._reset() + return line + + +def _printTiming(line: str) -> None: + import sys + from datetime import datetime + + stamp = datetime.now().strftime("%H:%M:%S.%f")[:-3] + print(f"{stamp} [lightsheet] {line}", file=sys.stderr, flush=True) + + # --------------------------------------------------------------------------- # depends_on / doc discovery @@ -282,6 +418,7 @@ def __init__(self, session, workers: int = 1, max_workers: int | None = None): self._direct_hits = 0 self._direct_fetches = 0 self._fallbacks = 0 + self.timing = _TileTiming(self) @staticmethod def _fileDir(session) -> str | None: @@ -1053,7 +1190,6 @@ def levelArrays( label per step. """ import dask.array as da - from dask import delayed from ndi.gui.app.lightsheetZarr import progress @@ -1119,21 +1255,21 @@ def levelArrays( fallback_reader = _makeFallbackReader(p, fallback_context) with progress.stage(f"preparing level {i} ({n_blocks:,} tiles, shape={list(shape)})"): - nested = _build_block_grid( - fetcher, - doc, - shape, - chunks, - chunk_grid, - dtype, - fill, - codec, - stored, - delayed, - da, - fallback_reader=fallback_reader, + arrays.append( + _build_level_array( + fetcher, + doc, + shape, + chunks, + chunk_grid, + dtype, + fill, + codec, + stored, + da, + fallback_reader=fallback_reader, + ) ) - arrays.append(da.block(nested)) return arrays, fetcher @@ -1171,7 +1307,7 @@ def _prepFallbackContext(docs: list) -> dict | None: def _makeFallbackReader(fine_props: dict, ctx: dict): """Return a closure over the fallback context for one fine level. - The closure has the signature ``_build_block_grid`` expects for + The closure has the signature ``_build_level_array`` expects for ``fallback_reader``: ``(fetcher, doc, filename, indices, chunks, block_shape, dtype, fill, codec)``. It reads the current refresh hint from ``ctx["refresh_hint_slot"][0]`` per call, so a hint @@ -1487,7 +1623,7 @@ def _linear_chunk_index(indices: tuple, chunk_grid: tuple) -> int: return linear + 1 -def _build_block_grid( +def _build_level_array( fetcher, doc, shape, @@ -1497,19 +1633,29 @@ def _build_block_grid( fill, codec, stored, - delayed, da, *, fallback_reader=None, ): - """Recursively build a nested list of dask blocks matching chunk_grid. - - Chunk paths are resolved LAZILY, inside each delayed task, through - the shared :class:`_ChunkFetcher`. That is what makes the reader - stream on a cloud-backed session: nothing is fetched until dask asks - for a specific block, and each block goes through fetcher threads - that own their own session handle (dodging DID's thread-owned - SQLite connection). + """One lazy dask array for a level, one block per chunk file. + + The whole level is ONE graph layer (``da.map_blocks`` with no input + arrays; each block learns its chunk position from ``block_info``). + An earlier version built one ``from_delayed`` array per chunk and + joined them with ``da.block``: every chunk became its own graph + layer, and dask walks every layer each time a slice is computed. A + view of 16 cached chunks then cost ~1.4 s on a 16k-chunk level and + ~26 s on a 131k-chunk one, against ~15 ms to read the 16 files -- + the cost grew with the size of the level, not of the view. A single + layer costs only for the blocks a slice touches, and the level is + built in microseconds rather than seconds. + + Chunk paths are resolved LAZILY, inside each block, through the + shared :class:`_ChunkFetcher`. That is what makes the reader stream + on a cloud-backed session: nothing is fetched until dask asks for a + specific block, and each block goes through fetcher threads that own + their own session handle (dodging DID's thread-owned SQLite + connection). STORED (a set of ``chunk.bin_#`` names, or None) short-circuits the fetcher for chunks the level document doesn't list. On a sparse @@ -1517,7 +1663,7 @@ def _build_block_grid( network call at all. None means the level document has no file list, and the reader falls back to attempting every position. - ``fallback_reader`` swaps the delayed body from + ``fallback_reader`` swaps the block body from :func:`_read_chunk_from_fetcher` to a wrapper that shows coarser-level data (upsampled) when the fine chunk is not on disk yet, so unfinished tiles do not draw as black holes. Its signature @@ -1526,35 +1672,41 @@ def _build_block_grid( per-axis 0-based multi-index, which the wrapper needs to compute the covering coarse region. """ + import numpy as np + from dask.base import tokenize - def one_block(indices: tuple): - idx_1 = _linear_chunk_index(indices, chunk_grid) - # Edge blocks are smaller along one or more axes. - block_shape = tuple( - min(chunks[a], shape[a] - indices[a] * chunks[a]) for a in range(len(indices)) - ) - name = f"chunk.bin_{idx_1:d}" - # Short-circuit missing chunks BEFORE they reach a dask task, so - # sparse regions don't even schedule work. + chunks = tuple(int(c) for c in chunks) + chunk_grid = tuple(int(g) for g in chunk_grid) + block_chunks = tuple( + tuple(min(chunks[a], shape[a] - i * chunks[a]) for i in range(chunk_grid[a])) + for a in range(len(chunk_grid)) + ) + + def read_block(block_info=None): + info = block_info[None] + indices = tuple(int(i) for i in info["chunk-location"]) + block_shape = tuple(int(n) for n in info["chunk-shape"]) + name = f"chunk.bin_{_linear_chunk_index(indices, chunk_grid):d}" if stored is not None and name not in stored: - d = delayed(_zero_block)(block_shape, dtype, fill) - elif fallback_reader is not None: - d = delayed(fallback_reader)( + return _zero_block(block_shape, dtype, fill) + if fallback_reader is not None: + return fallback_reader( fetcher, doc, name, indices, chunks, block_shape, dtype, fill, codec ) - else: - d = delayed(_read_chunk_from_fetcher)( - fetcher, doc, name, chunks, block_shape, dtype, fill, codec - ) - return da.from_delayed(d, shape=block_shape, dtype=dtype) + return _read_chunk_from_fetcher(fetcher, doc, name, chunks, block_shape, dtype, fill, codec) - def recurse(prefix: list) -> list: - axis = len(prefix) - if axis == len(chunk_grid): - return one_block(tuple(prefix)) - return [recurse(prefix + [i]) for i in range(chunk_grid[axis])] - - return recurse([]) + # An explicit name: tokenizing the closure would hash the fetcher. + # id(fetcher) keeps two viewers on the same level from sharing keys. + token = tokenize( + getattr(doc, "id", str(doc)), shape, chunks, str(dtype), fill, codec, id(fetcher) + ) + return da.map_blocks( + read_block, + chunks=block_chunks, + dtype=dtype, + meta=np.empty((0,) * len(block_chunks), dtype=dtype), + name=f"ndi-lightsheet-level-{token}", + ) def _zero_block(block_shape, dtype, fill): @@ -1579,9 +1731,17 @@ def _read_chunk_from_fetcher(fetcher, doc, filename, chunks_full, block_shape, d """ import numpy as np + t0 = time.monotonic() + key = (getattr(doc, "id", str(doc)), filename) + paths = getattr(fetcher, "_paths", None) + memo_hit = paths is not None and key in paths path = fetcher.chunkPath(doc, filename) + t_path = time.monotonic() n_expected = int(np.prod(chunks_full)) * dtype.itemsize + timing = getattr(fetcher, "timing", None) if path is None: + if timing is not None: + timing.record(t0, memo_hit, t_path - t0, False, 0.0, 0, 0.0) return np.full(block_shape, fill, dtype=dtype) try: raw = _readAll(path) @@ -1594,6 +1754,8 @@ def _read_chunk_from_fetcher(fetcher, doc, filename, chunks_full, block_shape, d raw = _readAll(path) except OSError: return np.full(block_shape, fill, dtype=dtype) + t_read = time.monotonic() + nbytes = len(raw) if codec == "blosc-zstd": from numcodecs import Blosc @@ -1605,6 +1767,10 @@ def _read_chunk_from_fetcher(fetcher, doc, filename, chunks_full, block_shape, d arr = np.frombuffer(raw[:n_expected], dtype=dtype).reshape(chunks_full) if block_shape != chunks_full: arr = arr[tuple(slice(0, s) for s in block_shape)] + if timing is not None: + timing.record( + t0, memo_hit, t_path - t0, True, t_read - t_path, nbytes, time.monotonic() - t_read + ) return arr diff --git a/src/ndi/pyramid/upsample_fallback.py b/src/ndi/pyramid/upsample_fallback.py index cd8b1311..e2db22c4 100644 --- a/src/ndi/pyramid/upsample_fallback.py +++ b/src/ndi/pyramid/upsample_fallback.py @@ -29,7 +29,7 @@ black" not "publication quality". On by default; opt out with ``NDI_LIGHTSHEET_UPSAMPLE_FALLBACK=0``. -When on, the reader path in :func:`multiscale._build_block_grid` +When on, the reader path in :func:`multiscale._build_level_array` routes each block through :func:`readChunkWithFallback` instead of :func:`multiscale._read_chunk_from_fetcher`. Reader also fires an asynchronous fine-chunk fetch (via ``fetcher.prefetchAsync``) so a From 37a9a6613d1c7166381c0ae934b6be5549c7ef72 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 23 Oct 2026 18:03:15 +0000 Subject: [PATCH 2/4] Gene pyramid: build each level as one dask layer levelArrays joined one from_delayed array per tile with da.block, so dask walked one graph layer per tile on every slice and drawing a view cost time in proportion to the whole level. Same fix as the lightsheet loader: one map_blocks layer per level, cropped to the true size. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EJ9PUeM7x2vLsg8FcZQfXo --- CHANGELOG.md | 2 ++ src/ndi/gui/app/genepyramid/multiscale.py | 35 ++++++++++++++++------- 2 files changed, 26 insertions(+), 11 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index dbe7447f..00a2e17e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -115,6 +115,8 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). The viewer also now prints one `[lightsheet] tiles:` line per burst of reads showing where tile time goes (path lookup, queue wait, disk read, decode, and the burst's wall-clock span); `NDI_LIGHTSHEET_TIMING=0` turns it off. + The gene-expression pyramid viewer built its levels the same way and gets + the same fix. - **Saving preferences no longer drops NDI-matlab's own entries.** `~/.ndi/NDI_Preferences.json` is shared with NDI-matlab, and each side diff --git a/src/ndi/gui/app/genepyramid/multiscale.py b/src/ndi/gui/app/genepyramid/multiscale.py index 88d141f1..5addfb92 100644 --- a/src/ndi/gui/app/genepyramid/multiscale.py +++ b/src/ndi/gui/app/genepyramid/multiscale.py @@ -275,7 +275,9 @@ def levelArrays(session, pyr_doc, gene_rows=None, density: bool = True) -> list[ Returns: A list of dask arrays, finest first. """ - dask, da = _require_dask() + _dask, da = _require_dask() + from dask.base import tokenize + levels, _frame = levelTable(session, pyr_doc) fetcher = _TileFetcher(session) @@ -303,8 +305,7 @@ def levelArrays(session, pyr_doc, gene_rows=None, density: bool = True) -> list[ # The block being lazy never helped, because the cost was already # paid by the time dask chose its blocks. def block( - r, - c, + block_info=None, _fetch=fetcher, _doc=tile_doc, _cols=cols, @@ -315,6 +316,7 @@ def block( _name=tileFileName, _lv=lv_props, ): + r, c = block_info[None]["chunk-location"] name = _name(_lv, r * _cols + c) if name not in _stored: return np.zeros((_th, _tw), np.float32) # missing tile costs nothing @@ -330,13 +332,24 @@ def block( tile = readTileFile(_fetch.tilePath(_doc, name)) return renderTile(tile, gene_rows, _th, _tw, binSize=_scale) - grid = [ - [ - da.from_delayed(dask.delayed(block)(r, c), shape=(th, tw), dtype=np.float32) - for c in range(cols) - ] - for r in range(rows) - ] + # ONE graph layer per level, one block per tile. Building the level + # as one from_delayed array per tile joined by da.block gave dask one + # layer per tile, and dask walks every layer each time napari + # computes a slice -- so drawing a view cost time in proportion to + # the whole level, not to the tiles on screen (seconds on a level of + # tens of thousands of tiles). The name is explicit: tokenizing the + # closure would hash the fetcher, and id(fetcher) keeps two viewers + # of the same pyramid from sharing keys. + token = tokenize( + getattr(tile_doc, "id", str(tile_doc)), b, gene_rows, density, th, tw, id(fetcher) + ) + level = da.map_blocks( + block, + chunks=((th,) * rows, (tw,) * cols), + dtype=np.float32, + meta=np.empty((0, 0), np.float32), + name=f"ndi-genepyramid-level-{token}", + ) # Crop to the level's TRUE size. The assembled grid is # rows*th by cols*tw, which overshoots whenever the level does not # divide evenly by the grid. Leaving the padding on makes a level @@ -344,7 +357,7 @@ def block( # pyramid disagree about their own field of view and a viewer # registering them by shape ratio misaligns them -- worse at coarse # levels, where the rounding is proportionally larger. - out.append(da.block(grid)[:lh, :lw]) + out.append(level[:lh, :lw]) return out From b49c98f7e54e0b7c9c4643ef8e7470d6b36abad8 Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 23 Oct 2026 18:45:12 +0000 Subject: [PATCH 3/4] Lightsheet: store fetched tiles in DID's cache without scanning its index A fetched tile took about a minute to be added to DID's file cache once it held tens of thousands of files and many download threads finished together: _keepFetched asked the index (is_file, a row-by-row scan), and DID-python's lock let threads through unguarded or made them sleep a second per retry. Check the disk instead, and let one fetcher thread at a time add to the cache. The debug fetch line now reports the add time. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EJ9PUeM7x2vLsg8FcZQfXo --- CHANGELOG.md | 7 +++ src/ndi/pyramid/multiscale.py | 39 ++++++++++++---- tests/test_pyramid_fetch_direct.py | 73 ++++++++++++++++++++++++++++++ 3 files changed, 109 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 00a2e17e..f2ab6427 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -117,6 +117,13 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/). and the burst's wall-clock span); `NDI_LIGHTSHEET_TIMING=0` turns it off. The gene-expression pyramid viewer built its levels the same way and gets the same fix. + Downloaded tiles also took about a minute each to be stored in DID's file + cache: with dozens of download threads finishing together, each scanned the + cache's index row by row and waited on its lock a second at a time. The + viewer now checks the disk instead of the index and lets one thread at a + time add to the cache; the debug fetch line reports the add as + `cache add Xs`. The underlying cache fixes are VH-Lab/DID-python#89 and + VH-Lab/DID-matlab#221. - **Saving preferences no longer drops NDI-matlab's own entries.** `~/.ndi/NDI_Preferences.json` is shared with NDI-matlab, and each side diff --git a/src/ndi/pyramid/multiscale.py b/src/ndi/pyramid/multiscale.py index 55055a16..f71c2986 100644 --- a/src/ndi/pyramid/multiscale.py +++ b/src/ndi/pyramid/multiscale.py @@ -58,6 +58,14 @@ class StaleChunkRead(Exception): """ +# One fetcher thread at a time adds to DID's file cache. The cache is one +# object per process with one lock, and DID-python before VH-Lab/DID-python#89 +# let a second thread through that lock unguarded (its has_lock flag is +# shared) and made a thread that did meet it sleep a whole second per retry; +# with dozens of download threads finishing together, adding a tile took +# about a minute. Waiting here costs only the add itself. +_CACHE_ADD_LOCK = threading.Lock() + # Returned by _ChunkFetcher._resolveDirect when the session-free path cannot # answer (no manifest on disk yet, no cloud location, not a series member) and # the caller should go through the session's database_openbinarydoc instead. @@ -645,6 +653,7 @@ def _resolve(self, doc, filename) -> str | None: import time as _time t0 = _time.monotonic() + self._local.keep_s = None if self._direct: try: direct = self._resolveDirect(doc, filename) @@ -737,6 +746,9 @@ def _recordResolve(self, filename, dt: float, path) -> None: f"({'batch' if t.get('batch') else 'per-file'}), " f"transfer {xfer:.2f}s, {size / 1e6:.2f} MB{rate}" ) + keep = getattr(self._local, "keep_s", None) + if keep is not None: + detail += f", cache add {keep:.2f}s" else: detail = "no cloud timing recorded" print( @@ -868,7 +880,11 @@ def _fetchMember(self, doc_id, filename, stem, member_uid, manifest_location) -> raise FileNotFoundError( f"the cloud handler produced no file for {filename} (uid {member_uid})" ) - return self._keepFetched(scratch, member_uid) + t_keep = time.monotonic() + try: + return self._keepFetched(scratch, member_uid) + finally: + self._local.keep_s = time.monotonic() - t_keep except BaseException: try: os.remove(scratch) @@ -885,17 +901,20 @@ def _keepFetched(self, scratch: str, member_uid: str) -> str: cache = get_cache() if len(member_uid) == cache.file_name_characters: - if cache.is_file(member_uid): - existing = cache.full_path(member_uid) - if os.path.isfile(existing): - # Another fetch of the same uid won the race; same bytes. - os.remove(scratch) - return existing - cache.remove_file(member_uid) - cache.add_file(scratch, member_uid) + existing = cache.full_path(member_uid) + # On disk is what counts, and checking the disk is one stat; + # asking the cache's index (is_file) scanned it row by row. + # add_file itself drops an index row whose file is gone. + if os.path.isfile(existing): + # Another fetch of the same uid won the race; same bytes. + os.remove(scratch) + return existing + with _CACHE_ADD_LOCK: + if not os.path.isfile(existing): + cache.add_file(scratch, member_uid) if os.path.exists(scratch): os.remove(scratch) - return cache.full_path(member_uid) + return existing except (OSError, ValueError, struct.error): pass if self._file_dir: diff --git a/tests/test_pyramid_fetch_direct.py b/tests/test_pyramid_fetch_direct.py index 1b087f82..191b58c2 100644 --- a/tests/test_pyramid_fetch_direct.py +++ b/tests/test_pyramid_fetch_direct.py @@ -295,3 +295,76 @@ def test_level_arrays_start_max_threads_with_default_limit(monkeypatch): assert multiscale._default_workers() == 16 assert set(multiscale.WORKER_CHOICES) >= {8, 16, 32, 64} assert multiscale.MAX_WORKERS >= max(multiscale.WORKER_CHOICES) + + +def test_fetched_members_from_many_threads_all_land_in_the_cache(monkeypatch, tmp_path): + # Dozens of download threads finish together and each adds its member + # to DID's one, process-wide file cache. Every member must land, and + # the add must not ask the index whether the file is there (a row-by- + # row scan) when one stat of the disk answers it. + import did.common + from did.file import FileCache + + cache_dir = tmp_path / "filecache" + cache_dir.mkdir() + cache = FileCache(str(cache_dir), 33) + monkeypatch.setattr(did.common, "get_cache", lambda *a, **k: cache) + + def no_index_scan(self, name): + raise AssertionError("is_file scans the index; check the disk instead") + + monkeypatch.setattr(FileCache, "is_file", no_index_scan) + + fetcher = _ChunkFetcher(object()) + n = 16 + uids = [f"{i:033d}" for i in range(n)] + scratches = [] + for i, uid in enumerate(uids): + p = tmp_path / f"{uid}.part" + p.write_bytes(_content(i)) + scratches.append(str(p)) + start = threading.Barrier(n) + results, errors = {}, [] + + def keep(i): + try: + start.wait() + results[i] = fetcher._keepFetched(scratches[i], uids[i]) + except Exception as exc: # noqa: BLE001 - reported below + errors.append(exc) + + threads = [threading.Thread(target=keep, args=(i,)) for i in range(n)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert errors == [] + for i, uid in enumerate(uids): + assert results[i] == cache.full_path(uid) + with open(results[i], "rb") as f: + assert f.read() == _content(i) + assert not os.path.exists(scratches[i]) + names, _sizes, _ = cache.file_list() + assert sorted(names) == uids + + +def test_a_member_already_in_the_cache_is_not_added_twice(monkeypatch, tmp_path): + import did.common + from did.file import FileCache + + cache_dir = tmp_path / "filecache" + cache_dir.mkdir() + cache = FileCache(str(cache_dir), 33) + monkeypatch.setattr(did.common, "get_cache", lambda *a, **k: cache) + uid = "1" * 33 + first = tmp_path / "first.part" + first.write_bytes(b"same bytes") + fetcher = _ChunkFetcher(object()) + path = fetcher._keepFetched(str(first), uid) + second = tmp_path / "second.part" + second.write_bytes(b"same bytes") + assert fetcher._keepFetched(str(second), uid) == path + assert not second.exists() + names, _sizes, _ = cache.file_list() + assert names == [uid] From c2c14e9bb7c8473aa5b3aee0cb720b1bee5169cd Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 23 Oct 2026 19:35:39 +0000 Subject: [PATCH 4/4] Cloud downloads: pool a connection per lightsheet worker The shared download Session kept requests' default 10 connections per host while the lightsheet fetcher runs up to 64 downloads at once, so every download past the tenth opened its own connection and discarded it afterwards ("Connection pool is full, discarding connection"). Mount an adapter pooling 64. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01EJ9PUeM7x2vLsg8FcZQfXo --- src/ndi/cloud/api/files.py | 19 +++++++++++++++++-- tests/test_pyramid_fetch_direct.py | 12 ++++++++++++ 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/src/ndi/cloud/api/files.py b/src/ndi/cloud/api/files.py index fca513f2..746b1979 100644 --- a/src/ndi/cloud/api/files.py +++ b/src/ndi/cloud/api/files.py @@ -260,8 +260,11 @@ def _download_session(): connection across sibling ``getFile`` calls. Every tile of a pyramid goes to the same S3 host, so a shared Session amortises the TLS handshake (~100-500ms) across N chunks. Thread-safe for - GET; the default pool holds 10 connections which is comfortable - for the lightsheet fetcher's 8 workers. + GET. The pool holds ``_DOWNLOAD_POOL_SIZE`` connections per host: + the lightsheet fetcher runs up to ``ndi.pyramid.multiscale.MAX_WORKERS`` + (64) downloads at once, and with requests' default of 10 every + download past the tenth opened a connection of its own and threw it + away afterwards ("Connection pool is full, discarding connection"). A previous version opened a fresh connection per call (``requests.get(url, ...)``) which added handshake latency to @@ -271,11 +274,23 @@ def _download_session(): session = getattr(_download_session, "_session", None) if session is None: + from requests.adapters import HTTPAdapter + session = requests.Session() + adapter = HTTPAdapter( + pool_connections=_DOWNLOAD_POOL_SIZE, pool_maxsize=_DOWNLOAD_POOL_SIZE + ) + session.mount("https://", adapter) + session.mount("http://", adapter) _download_session._session = session return session +#: Connections kept per host by the download session; matches the most +#: downloads ndi.pyramid.multiscale.MAX_WORKERS runs at once. +_DOWNLOAD_POOL_SIZE = 64 + + def getFile( url: NonEmptyStr, target_path: str | Path, diff --git a/tests/test_pyramid_fetch_direct.py b/tests/test_pyramid_fetch_direct.py index 191b58c2..52416809 100644 --- a/tests/test_pyramid_fetch_direct.py +++ b/tests/test_pyramid_fetch_direct.py @@ -368,3 +368,15 @@ def test_a_member_already_in_the_cache_is_not_added_twice(monkeypatch, tmp_path) assert not second.exists() names, _sizes, _ = cache.file_list() assert names == [uid] + + +def test_the_download_session_pools_a_connection_per_worker(): + # Up to MAX_WORKERS tiles download at once through one Session. A pool + # smaller than that makes every extra download open, then discard, a + # connection of its own ("Connection pool is full"). + from ndi.cloud.api import files as files_api + + session = files_api._download_session() + adapter = session.get_adapter("https://example.s3.amazonaws.com/x") + assert files_api._DOWNLOAD_POOL_SIZE >= multiscale.MAX_WORKERS + assert adapter._pool_maxsize >= multiscale.MAX_WORKERS