diff --git a/CHANGELOG.md b/CHANGELOG.md index 8279d3b8..2424c6eb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -118,6 +118,27 @@ 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. + 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 rewrote it with only the preferences it registers, so saving from Python 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/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 diff --git a/src/ndi/pyramid/multiscale.py b/src/ndi/pyramid/multiscale.py index f3fd1af2..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. @@ -80,6 +88,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 +426,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: @@ -508,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) @@ -600,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( @@ -731,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) @@ -748,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: @@ -1053,7 +1209,6 @@ def levelArrays( label per step. """ import dask.array as da - from dask import delayed from ndi.gui.app.lightsheetZarr import progress @@ -1119,21 +1274,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 +1326,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 +1642,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 +1652,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 +1682,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 +1691,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) - - 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 _read_chunk_from_fetcher(fetcher, doc, name, chunks, block_shape, dtype, fill, codec) - 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 +1750,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 +1773,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 +1786,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 diff --git a/tests/test_pyramid_fetch_direct.py b/tests/test_pyramid_fetch_direct.py index 1b087f82..52416809 100644 --- a/tests/test_pyramid_fetch_direct.py +++ b/tests/test_pyramid_fetch_direct.py @@ -295,3 +295,88 @@ 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] + + +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