From 963180ae7c9e7f434ad2d00db5cb867eb0904b34 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Wed, 14 Oct 2026 13:38:26 -0700 Subject: [PATCH 1/5] perf(serverless): range-parallel VolumeCache hydrate Hydrate now splits every big file into 8 MiB ranges and copies the ranges of all files from one pool of 64 threads, each range on its own fds. Object-store mounts (Global Store's geesefs) serve one sequential stream at ~140 MiB/s, and whole-file parallelism doesn't add up. Worse, parallel whole-file copies there fail with ENOMEM, and hydrate dropped those files without retry. Measured on a 4090 pod in EU-RO-1, hydrating a 15.2 GB 7B model from a global volume: before: 9 of 11 files (2 of 4 weight shards lost to ENOMEM), 24-53s after: 11 of 11 files, byte-identical, 17-22s (670-844 MiB/s) ENOMEM/EIO/EAGAIN are retried per range with backoff. Files are assembled in a temp file and renamed only when every range copied, so a failure leaves the local file untouched. A mirror shorter than its manifest entry fails instead of restoring zero padding. Tunable via hydrate_workers and hydrate_chunk_size. Sync is unchanged. --- docs/serverless/volume_cache.md | 22 +- runpod/serverless/utils/rp_volume_cache.py | 221 +++++++++++++++--- .../test_utils/test_rp_volume_cache.py | 179 ++++++++++++++ 3 files changed, 382 insertions(+), 40 deletions(-) diff --git a/docs/serverless/volume_cache.md b/docs/serverless/volume_cache.md index d3ebebda9..1e9897e3a 100644 --- a/docs/serverless/volume_cache.md +++ b/docs/serverless/volume_cache.md @@ -59,7 +59,9 @@ vc.sync(background=False) # persist new files back to the volume, inli | `namespace` | `RUNPOD_ENDPOINT_ID` | Isolation key for the on-volume mirror. Must be a single safe path component. | | `volume_path` | `/runpod-volume` | Network-volume mount point. | | `best_effort` | `True` | Swallow and log errors instead of raising. Set `False` while debugging. | -| `max_workers` | `min(32, (os.cpu_count() or 4) * 4)` | Thread count for parallel copy of large files (I/O-bound). | +| `max_workers` | `min(32, (os.cpu_count() or 4) * 4)` | Thread count for parallel copy of large files on sync (I/O-bound). | +| `hydrate_workers` | `64` | Concurrent range reads when hydrating large files. | +| `hydrate_chunk_size` | `8 MiB` | Bytes per hydrate range. Peak hydrate memory is about `hydrate_workers * hydrate_chunk_size` (512 MiB by default). | ## How it works @@ -77,9 +79,21 @@ vc.sync(background=False) # persist new files back to the volume, inli files are skipped. The `small.tar` archive is re-packed as a whole whenever any small file has changed, since unpacking and re-diffing many tiny files individually is slower than the network volume's per-file overhead. -- **Parallel copy.** Large-file transfers run across a thread pool sized by - `max_workers` (default `min(32, (os.cpu_count() or 4) * 4)`), since the - work is I/O-bound. +- **Parallel copy.** On sync, large files are copied whole across a thread + pool sized by `max_workers` (default `min(32, (os.cpu_count() or 4) * 4)`), + since the work is I/O-bound. +- **Range-parallel hydrate.** On hydrate, every large file is split into + `hydrate_chunk_size` ranges, and one pool of `hydrate_workers` threads reads + the ranges of all files at once, each on its own file descriptor. + Object-store mounts such as Global Store serve one sequential stream at + roughly 140 MiB/s, and reading several whole files at once doesn't add up, + but independent range reads scale to the host's network (~1.1 GiB/s + measured). Transient `ENOMEM`/`EIO`/`EAGAIN` errors, which such mounts + return under load, are retried with backoff. A file is assembled in a temp + file and renamed into place only when every range has copied, so a failed + file leaves any existing local copy untouched and is simply not restored. + Keep `hydrate_workers * hydrate_chunk_size` near the default: past ~1 GiB in + flight, Global Store starts failing reads with `ENOMEM`. - **Packing/extraction.** Packing the small-file archive uses the `tar` binary when available (falling back to the stdlib `tarfile` module otherwise). Extraction always goes through `tarfile`, validating every diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index cd654eb65..ffeffd717 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -11,8 +11,11 @@ Transport adapts to the tree's shape: files below 256 KiB are packed into a single ``small.tar`` on the volume (collapsing per-file metadata round-trips), larger files are copied unpacked into ``big/`` across a thread pool, and a -versioned ``manifest.json`` (written last) records the layout. Large-file -transfers stay incremental (size/mtime diff); the small-file archive is +versioned ``manifest.json`` (written last) records the layout. Hydrate restores +large files with concurrent range reads instead of whole-file copies, which is +what object-store mounts such as Global Store need to go faster than one +sequential stream. Large-file transfers stay incremental (size/mtime diff); the +small-file archive is re-packed whole when any small file changes. Packing uses the ``tar`` binary when present (else stdlib ``tarfile``); extraction always uses ``tarfile``, validating every member against the configured dirs. Best-effort: any failure @@ -20,12 +23,15 @@ """ import atexit +import collections +import errno import json import os import shutil import subprocess import tarfile import threading +import time from concurrent.futures import ThreadPoolExecutor from runpod.serverless.modules.rp_logger import RunPodLogger @@ -39,6 +45,18 @@ _MANIFEST_NAME = "manifest.json" _SMALL_ARCHIVE_NAME = "small.tar" _BIG_SUBDIR = "big" +# Hydrate copies big files in ranges. Bytes in flight (workers x chunk, 512 MiB +# here) is what matters: on Global Store's geesefs mount, 64 x 8 MiB reaches the +# host NIC (~1.1 GiB/s), and much past ~1 GiB in flight fails reads with ENOMEM. +_HYDRATE_WORKERS = 64 +_HYDRATE_CHUNK_SIZE = 8 * 1024 * 1024 +# Transient errors an object-store FUSE mount returns under load; retried with +# exponential backoff (~6s in total) before the file is given up on. +_RANGE_RETRY_ERRNOS = frozenset({errno.ENOMEM, errno.EIO, errno.EAGAIN}) +_RANGE_MAX_ATTEMPTS = 8 +_RANGE_RETRY_BASE_SECONDS = 0.05 + +_RangedFile = collections.namedtuple("_RangedFile", "src dst tmp size mtime") class VolumeCache: @@ -62,9 +80,12 @@ class VolumeCache: volume_path: Network-volume mount point. Defaults to ``/runpod-volume``. best_effort: When True (default), swallow and log errors instead of raising. - max_workers: Thread count for parallel copy of large files. Defaults - to ``min(32, (os.cpu_count() or 4) * 4)`` (I/O-bound, so + max_workers: Thread count for parallel copy of large files on sync. + Defaults to ``min(32, (os.cpu_count() or 4) * 4)`` (I/O-bound, so oversubscribing the CPU count is fine). + hydrate_workers: Concurrent range reads on hydrate. Defaults to 64. + hydrate_chunk_size: Bytes per hydrate range. Defaults to 8 MiB. Peak + hydrate memory is about ``hydrate_workers * hydrate_chunk_size``. Example: >>> with VolumeCache(dirs=["/root/.cache/huggingface"]): @@ -74,7 +95,15 @@ class VolumeCache: _EXCLUDE_SUBSTRINGS = (os.sep + "refs" + os.sep, os.sep + ".no_exist" + os.sep) def __init__( - self, dirs, *, namespace=None, volume_path="/runpod-volume", best_effort=True, max_workers=None + self, + dirs, + *, + namespace=None, + volume_path="/runpod-volume", + best_effort=True, + max_workers=None, + hydrate_workers=None, + hydrate_chunk_size=None, ): if isinstance(dirs, (str, bytes, os.PathLike)): # A bare path is one directory, not an iterable of path characters: @@ -101,12 +130,13 @@ def __init__( # the cache. self._volume_path = os.fsdecode(os.fspath(volume_path)) self._best_effort = best_effort - # A non-positive max_workers is a caller bug: ThreadPoolExecutor would - # raise ValueError mid-sync, which best-effort mode swallows and silently - # disables syncing. Reject it up front like the namespace check above. - if max_workers is not None and max_workers < 1: - raise ValueError(f"max_workers must be a positive integer, got {max_workers!r}") - self._max_workers = max_workers or min(32, (os.cpu_count() or 4) * 4) + self._max_workers = _positive_or_default( + "max_workers", max_workers, min(32, (os.cpu_count() or 4) * 4) + ) + self._hydrate_workers = _positive_or_default("hydrate_workers", hydrate_workers, _HYDRATE_WORKERS) + self._hydrate_chunk_size = _positive_or_default( + "hydrate_chunk_size", hydrate_chunk_size, _HYDRATE_CHUNK_SIZE + ) @property def _mirror_root(self): @@ -173,7 +203,7 @@ def _write_manifest(self, small_meta, big_meta): "small": small_meta, "big": big_meta, } - tmp = f"{self._manifest_path}.{os.getpid()}.{threading.get_ident()}.rpvc.tmp" + tmp = _tmp_path(self._manifest_path) with open(tmp, "w") as fh: json.dump(payload, fh) os.replace(tmp, self._manifest_path) @@ -252,12 +282,12 @@ def _is_safe_dest(self, dst_abs): target = os.path.realpath(dst_abs) return any(target == d or target.startswith(d + os.sep) for d in self._dirs) - def _copy_file(self, src, dst, mtime=None): - tmp = f"{dst}.{os.getpid()}.{threading.get_ident()}.rpvc.tmp" + def _copy_file(self, src, dst): + tmp = _tmp_path(dst) try: os.makedirs(os.path.dirname(dst), exist_ok=True) shutil.copyfile(src, tmp) - _copy_metadata(src, tmp, mtime) + _copy_metadata(src, tmp) os.replace(tmp, dst) # atomic; last-writer-wins under concurrency return True except OSError as exc: @@ -270,12 +300,11 @@ def _copy_file(self, src, dst, mtime=None): pass return False - def _copy_parallel(self, pairs, needs_copy=None, mtime_of=None): + def _copy_parallel(self, pairs, needs_copy=None): """Copy every (src, dst) that needs it, in parallel. ``needs_copy(src, dst)`` decides which pairs to copy (default: compare - the two files). ``mtime_of(src)`` gives the mtime to stamp on dst - (default: src's own). + the two files). Returns ``(copied, current)`` sets of source paths: - ``copied`` -- sources transferred successfully this call. @@ -289,7 +318,6 @@ def _copy_parallel(self, pairs, needs_copy=None, mtime_of=None): at that instant, so no re-stat is needed to confirm it. """ needs_copy = needs_copy or self._needs_copy - mtime_of = mtime_of or (lambda _src: None) todo = [(s, d) for s, d in pairs if needs_copy(s, d)] current = {s for s, _ in pairs} - {s for s, _ in todo} if not todo: @@ -298,10 +326,79 @@ def _copy_parallel(self, pairs, needs_copy=None, mtime_of=None): # pool.map preserves input order, so zip pairs each result with its # source. This holds one path per copied file (unavoidable: per-file # identity is required to keep failed copies out of the manifest). - results = pool.map(lambda sd: self._copy_file(sd[0], sd[1], mtime=mtime_of(sd[0])), todo) + results = pool.map(lambda sd: self._copy_file(sd[0], sd[1]), todo) copied = {s for (s, _), ok in zip(todo, results) if ok} return copied, current + def _hydrate_ranged(self, entries): + """Restore ``(src, dst, manifest_entry)`` triples with concurrent range reads. + + Every file is split into ``hydrate_chunk_size`` ranges, and one pool of + ``hydrate_workers`` threads copies the ranges of all files, each range on + its own descriptors. An object-store mount serves one sequential stream + at ~140 MiB/s however many whole files are read at once, but independent + range reads scale to the host NIC. A file is written to a temp file and + renamed into place only once every range copied, stamped with the + manifest's source mtime, so a failure leaves the local file untouched. + + Returns the set of sources restored. + """ + files = [f for f in (self._prepare_ranged(*e) for e in entries) if f] + failed = self._copy_ranges(files) + return {f.src for i, f in enumerate(files) if self._finish_ranged(f, i in failed)} + + def _prepare_ranged(self, src, dst, entry): + tmp = _tmp_path(dst) + try: + os.makedirs(os.path.dirname(dst), exist_ok=True) + with open(tmp, "wb") as fh: + fh.truncate(entry["size"]) + except OSError as exc: + log.warn(f"VolumeCache: failed to hydrate {src} -> {dst}: {exc}") + _silent_remove(tmp) + return None + return _RangedFile(src, dst, tmp, entry["size"], entry["mtime"]) + + def _copy_ranges(self, files): + """Copy every range of ``files`` into their temp files; return indexes that failed.""" + chunk = self._hydrate_chunk_size + work = [ + (off, i, min(chunk, f.size - off)) for i, f in enumerate(files) for off in range(0, f.size, chunk) + ] + # Offset-major order spreads concurrent readers across every file. + work.sort(key=lambda w: (w[0], files[w[1]].src)) + failed = set() + + def run(item): + off, i, length = item + if i in failed: + return + f = files[i] + try: + _with_range_retry(_copy_range, f.src, f.tmp, off, length) + except OSError as exc: + failed.add(i) + log.warn(f"VolumeCache: failed to hydrate {f.src} at offset {off}: {exc}") + + if work: + with ThreadPoolExecutor(max_workers=self._hydrate_workers) as pool: + list(pool.map(run, work)) + return failed + + @staticmethod + def _finish_ranged(f, failed): + if failed: + _silent_remove(f.tmp) + return False + try: + _copy_metadata(f.src, f.tmp, f.mtime) + os.replace(f.tmp, f.dst) + return True + except OSError as exc: + log.warn(f"VolumeCache: failed to hydrate {f.src} -> {f.dst}: {exc}") + _silent_remove(f.tmp) + return False + @staticmethod def _within(base, path): """True iff ``path`` resolves inside ``base`` (symlinks resolved).""" @@ -323,7 +420,7 @@ def _tar_binary(self): def _pack_small(self, files): os.makedirs(self._mirror_root, exist_ok=True) - tmp = f"{self._small_archive_path}.{os.getpid()}.{threading.get_ident()}.rpvc.tmp" + tmp = _tmp_path(self._small_archive_path) rels = [os.path.relpath(f, "/") for f in files] try: if self._tar_binary(): @@ -418,27 +515,24 @@ def _do_hydrate(self): restored = 0 if manifest.get("small"): restored += self._extract_small() - big_pairs = [] - entries = {} + todo = [] for entry in manifest.get("big", []): rel = os.path.relpath(entry["path"], "/") dst = os.path.join("/", rel) src = os.path.join(self._big_root, rel) # Guard both ends against a crafted manifest: the destination must # stay inside a configured dir, and the source must resolve inside - # big/ so a malicious entry can't read outside the mirror. - if self._is_safe_dest(dst) and self._within(self._big_root, src): - big_pairs.append((src, dst)) - entries[src] = entry - # Compare the container against the manifest's source metadata, not - # the mirror file's own mtime, and stamp that source mtime on restore. - copied, _current = self._copy_parallel( - big_pairs, - needs_copy=lambda src, _dst: os.path.isfile(src) - and not self._meta_satisfied_by_local(entries[src]), - mtime_of=lambda src: entries[src]["mtime"], - ) - restored += len(copied) + # big/ so a malicious entry can't read outside the mirror. Compare + # the container against the manifest's source metadata, not the + # mirror file's own mtime. + if ( + self._is_safe_dest(dst) + and self._within(self._big_root, src) + and os.path.isfile(src) + and not self._meta_satisfied_by_local(entry) + ): + todo.append((src, dst, entry)) + restored += len(self._hydrate_ranged(todo)) if restored: log.info(f"VolumeCache: hydrated {restored} file(s) from {self._mirror_root}") return restored @@ -558,6 +652,61 @@ def _copy_metadata(src, dst, mtime=None): log.debug(f"VolumeCache: kept default mode on {dst}: {exc}") +def _positive_or_default(name, value, default): + # A non-positive pool size or chunk is a caller bug that best-effort mode + # would otherwise swallow mid-operation, silently disabling the cache. + if value is not None and value < 1: + raise ValueError(f"{name} must be a positive integer, got {value!r}") + return value or default + + +def _tmp_path(path): + return f"{path}.{os.getpid()}.{threading.get_ident()}.rpvc.tmp" + + +def _with_range_retry(fn, *args): + """Call fn, retrying transient FUSE errors with exponential backoff.""" + for attempt in range(_RANGE_MAX_ATTEMPTS): + try: + return fn(*args) + except OSError as exc: + if exc.errno not in _RANGE_RETRY_ERRNOS or attempt == _RANGE_MAX_ATTEMPTS - 1: + raise + time.sleep(_RANGE_RETRY_BASE_SECONDS * 2**attempt) + + +def _copy_range(src, dst, offset, length): + """Copy ``length`` bytes at ``offset`` of src to the same offset of dst. + + Each call opens its own descriptors so an object-store mount serves the + range as an independent read. A short read means the mirror file is + smaller than the manifest says; it raises without an errno, so it is not + retried. + """ + src_fd = os.open(src, os.O_RDONLY) + try: + dst_fd = os.open(dst, os.O_WRONLY) + try: + done = 0 + while done < length: + buf = os.pread(src_fd, length - done, offset + done) + if not buf: + raise OSError(f"short read: {done} of {length} bytes at offset {offset}") + _pwrite_all(dst_fd, buf, offset + done) + done += len(buf) + finally: + os.close(dst_fd) + finally: + os.close(src_fd) + + +def _pwrite_all(fd, buf, offset): + view = memoryview(buf) + written = 0 + while written < len(view): + written += os.pwrite(fd, view[written:], offset + written) + + def _silent_remove(path): try: os.remove(path) diff --git a/tests/test_serverless/test_utils/test_rp_volume_cache.py b/tests/test_serverless/test_utils/test_rp_volume_cache.py index 179eb1a9e..492dce118 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -1088,3 +1088,182 @@ def test_copy_parallel_copies_needed_only(tmp_path): def test_copy_parallel_empty_is_zero(tmp_path): vc, _cache, _vol = _mk_cache_with_volume(tmp_path) assert vc._copy_parallel([]) == (set(), set()) + + +# --------------------------------------------------------------------------- # +# range-parallel hydrate +# --------------------------------------------------------------------------- # + +_CHUNK = 64 * 1024 + + +def _synced_big_files(tmp_path, monkeypatch, names, size=None): + """Sync big files to the volume, delete them locally; return (cache, vol, payloads).""" + monkeypatch.setattr(vcmod, "_RANGE_RETRY_BASE_SECONDS", 0) + vc, cache, vol = _mk_cache_with_volume(tmp_path) + size = size or vcmod._SMALL_FILE_THRESHOLD * 3 + 123 # not a multiple of the chunk + payloads = {} + for name in names: + payloads[name] = os.urandom(size) + (cache / name).write_bytes(payloads[name]) + vc.sync(background=False) + for name in names: + (cache / name).unlink() + return cache, vol, payloads + + +def _hydrating_cache(cache, vol, **kwargs): + kwargs.setdefault("hydrate_chunk_size", _CHUNK) + return VolumeCache([str(cache)], namespace="ep1", volume_path=str(vol), **kwargs) + + +def _fail_pread_on(monkeypatch, marker, err, times=None): + """Make os.pread raise OSError(err) for fds of files whose path contains marker.""" + real_open, real_pread = os.open, os.pread + marked, calls = set(), [] + + def tracking_open(path, flags, *args, **kwargs): + fd = real_open(path, flags, *args, **kwargs) + if marker in str(path) and not flags & os.O_WRONLY: + marked.add(fd) + else: + marked.discard(fd) + return fd + + def flaky_pread(fd, n, off): + if fd in marked and (times is None or len(calls) < times): + calls.append(off) + raise OSError(err, os.strerror(err)) + return real_pread(fd, n, off) + + monkeypatch.setattr(os, "open", tracking_open) + monkeypatch.setattr(os, "pread", flaky_pread) + return calls + + +def test_hydrate_copies_big_file_in_ranges(tmp_path, monkeypatch): + cache, vol, payloads = _synced_big_files(tmp_path, monkeypatch, ["model.safetensors"]) + ranges = [] + real_copy_range = vcmod._copy_range + monkeypatch.setattr( + vcmod, "_copy_range", lambda *a: ranges.append(a[2:]) or real_copy_range(*a) + ) + + restored = _hydrating_cache(cache, vol).hydrate() + + size = len(payloads["model.safetensors"]) + assert restored == 1 + assert (cache / "model.safetensors").read_bytes() == payloads["model.safetensors"] + assert sorted(ranges) == [(off, min(_CHUNK, size - off)) for off in range(0, size, _CHUNK)] + + +def test_hydrate_interleaves_ranges_across_files(tmp_path, monkeypatch): + # Concurrent readers must spread over every file: on an object-store mount, + # parallel reads of one sequential stream don't add up. + cache, vol, _payloads = _synced_big_files(tmp_path, monkeypatch, ["a.bin", "b.bin"]) + order = [] + real_copy_range = vcmod._copy_range + monkeypatch.setattr( + vcmod, + "_copy_range", + lambda src, *a: order.append(os.path.basename(src)) or real_copy_range(src, *a), + ) + + _hydrating_cache(cache, vol, hydrate_workers=1).hydrate() + + assert order[:4] == ["a.bin", "b.bin", "a.bin", "b.bin"] + + +def test_hydrate_retries_transient_range_errors(tmp_path, monkeypatch): + cache, vol, payloads = _synced_big_files(tmp_path, monkeypatch, ["model.safetensors"]) + failures = _fail_pread_on(monkeypatch, "/big/", errno.ENOMEM, times=3) + + restored = _hydrating_cache(cache, vol).hydrate() + + assert len(failures) == 3 + assert restored == 1 + assert (cache / "model.safetensors").read_bytes() == payloads["model.safetensors"] + + +def test_hydrate_does_not_retry_non_transient_errors(tmp_path, monkeypatch): + cache, vol, _payloads = _synced_big_files(tmp_path, monkeypatch, ["model.safetensors"]) + failures = _fail_pread_on(monkeypatch, "/big/", errno.EACCES) + + restored = _hydrating_cache(cache, vol, hydrate_workers=1).hydrate() + + assert restored == 0 + assert len(failures) == 1 # first failure aborts the file; its other ranges are skipped + + +def test_hydrate_skips_only_the_failing_file(tmp_path, monkeypatch): + cache, vol, payloads = _synced_big_files(tmp_path, monkeypatch, ["bad.bin", "good.bin"]) + _fail_pread_on(monkeypatch, "bad.bin", errno.EIO) + + restored = _hydrating_cache(cache, vol).hydrate() + + assert restored == 1 + assert (cache / "good.bin").read_bytes() == payloads["good.bin"] + assert not (cache / "bad.bin").exists() + assert [p for p in os.listdir(cache) if p.endswith(".rpvc.tmp")] == [] + + +def test_hydrate_failure_leaves_existing_local_file_untouched(tmp_path, monkeypatch): + cache, vol, _payloads = _synced_big_files(tmp_path, monkeypatch, ["model.safetensors"]) + stale = cache / "model.safetensors" + stale.write_bytes(b"older local bytes") + _set_mtime(stale, 1_000_000_000) + _fail_pread_on(monkeypatch, "/big/", errno.EIO) + + assert _hydrating_cache(cache, vol).hydrate() == 0 + assert stale.read_bytes() == b"older local bytes" + + +def test_hydrate_treats_short_read_as_failure(tmp_path, monkeypatch): + # A mirror file truncated after the manifest was written must not be + # restored as a zero-padded copy of the expected size. + cache, vol, _payloads = _synced_big_files(tmp_path, monkeypatch, ["model.safetensors"]) + mirrored = vol / ".cache" / "ep1" / "big" / os.path.relpath(str(cache / "model.safetensors"), "/") + with open(mirrored, "r+b") as fh: + fh.truncate(_CHUNK + 1) + + assert _hydrating_cache(cache, vol).hydrate() == 0 + assert not (cache / "model.safetensors").exists() + + +@pytest.mark.parametrize("arg", ["hydrate_workers", "hydrate_chunk_size"]) +@pytest.mark.parametrize("bad", [0, -1]) +def test_hydrate_tuning_rejects_non_positive(tmp_path, arg, bad): + vol = tmp_path / "volume" + vol.mkdir() + with pytest.raises(ValueError): + VolumeCache([str(tmp_path / "cache")], namespace="ep1", volume_path=str(vol), **{arg: bad}) + + +def test_hydrate_tuning_defaults(tmp_path): + vc, _cache, _vol = _mk_cache_with_volume(tmp_path) + assert vc._hydrate_workers == 64 + assert vc._hydrate_chunk_size == 8 * 1024 * 1024 + + +def test_hydrate_skips_file_when_local_temp_cannot_be_created(tmp_path, monkeypatch): + cache, vol, _payloads = _synced_big_files(tmp_path, monkeypatch, ["model.safetensors"]) + + def no_space(*_args, **_kwargs): + raise OSError(errno.ENOSPC, "No space left on device") + + monkeypatch.setattr(os, "makedirs", no_space) + + assert _hydrating_cache(cache, vol).hydrate() == 0 + assert os.listdir(cache) == [] + + +def test_hydrate_cleans_temp_when_rename_fails(tmp_path, monkeypatch): + cache, vol, _payloads = _synced_big_files(tmp_path, monkeypatch, ["model.safetensors"]) + + def busy(*_args, **_kwargs): + raise OSError(errno.EBUSY, "Device or resource busy") + + monkeypatch.setattr(os, "replace", busy) + + assert _hydrating_cache(cache, vol).hydrate() == 0 + assert os.listdir(cache) == [] From 51ffc0a8d207d9036dce3d0c9a3438d3e97e14df Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Wed, 14 Oct 2026 23:19:58 -0700 Subject: [PATCH 2/5] fix(serverless): harden ranged VolumeCache hydrate Address review feedback on #621: - Pin each mirror file's size+mtime before its first range and reject the file if a range opens a replaced source, or if the mirror size disagrees with the manifest (a sync that landed after the manifest was read), instead of publishing mixed or truncated bytes. - Generate ranges lazily and pull them from hydrate_workers threads so scheduling stays O(workers) instead of one future per range. - Stop remaining ranges and remove pre-sized temp files when an unexpected (non-OSError) error aborts the hydrate. - Reject non-integer and bool tuning values at construction. --- docs/serverless/volume_cache.md | 6 +- runpod/serverless/utils/rp_volume_cache.py | 118 ++++++++++++++---- .../test_utils/test_rp_volume_cache.py | 94 +++++++++++++- 3 files changed, 194 insertions(+), 24 deletions(-) diff --git a/docs/serverless/volume_cache.md b/docs/serverless/volume_cache.md index 1e9897e3a..cf79abb85 100644 --- a/docs/serverless/volume_cache.md +++ b/docs/serverless/volume_cache.md @@ -92,7 +92,11 @@ vc.sync(background=False) # persist new files back to the volume, inli return under load, are retried with backoff. A file is assembled in a temp file and renamed into place only when every range has copied, so a failed file leaves any existing local copy untouched and is simply not restored. - Keep `hydrate_workers * hydrate_chunk_size` near the default: past ~1 GiB in + A mirror file whose size differs from the manifest, or whose size or mtime + changes between ranges because a concurrent `sync()` replaced it, is + likewise not restored. This is a best-effort check: an object-store mount + can serve a replacement that lands in the middle of a single range read + without it being visible. Keep `hydrate_workers * hydrate_chunk_size` near the default: past ~1 GiB in flight, Global Store starts failing reads with `ENOMEM`. - **Packing/extraction.** Packing the small-file archive uses the `tar` binary when available (falling back to the stdlib `tarfile` module diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index ffeffd717..ccf6924d7 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -56,7 +56,7 @@ _RANGE_MAX_ATTEMPTS = 8 _RANGE_RETRY_BASE_SECONDS = 0.05 -_RangedFile = collections.namedtuple("_RangedFile", "src dst tmp size mtime") +_RangedFile = collections.namedtuple("_RangedFile", "src dst tmp size mtime identity") class VolumeCache: @@ -337,19 +337,35 @@ def _hydrate_ranged(self, entries): ``hydrate_workers`` threads copies the ranges of all files, each range on its own descriptors. An object-store mount serves one sequential stream at ~140 MiB/s however many whole files are read at once, but independent - range reads scale to the host NIC. A file is written to a temp file and + range reads scale to the host NIC. Each source's size and mtime are pinned + (and checked against the manifest) before its first range, and every + range rejects a source that a concurrent sync() visibly replaced since. + A file is written to a temp file and renamed into place only once every range copied, stamped with the manifest's source mtime, so a failure leaves the local file untouched. Returns the set of sources restored. """ files = [f for f in (self._prepare_ranged(*e) for e in entries) if f] - failed = self._copy_ranges(files) + try: + failed = self._copy_ranges(files) + except BaseException: + # Temps are pre-sized to the full file and named per pid, so a + # later run never reuses them; leaking them can fill the disk. + for f in files: + _silent_remove(f.tmp) + raise return {f.src for i, f in enumerate(files) if self._finish_ranged(f, i in failed)} def _prepare_ranged(self, src, dst, entry): tmp = _tmp_path(dst) try: + st = os.stat(src) + if st.st_size != entry["size"]: + # A sync replaced the mirror file after the manifest was read; + # copying entry["size"] bytes of it would publish a truncated file. + raise OSError(f"mirror size {st.st_size} != manifest size {entry['size']}") + identity = _file_identity(st) os.makedirs(os.path.dirname(dst), exist_ok=True) with open(tmp, "wb") as fh: fh.truncate(entry["size"]) @@ -357,16 +373,10 @@ def _prepare_ranged(self, src, dst, entry): log.warn(f"VolumeCache: failed to hydrate {src} -> {dst}: {exc}") _silent_remove(tmp) return None - return _RangedFile(src, dst, tmp, entry["size"], entry["mtime"]) + return _RangedFile(src, dst, tmp, entry["size"], entry["mtime"], identity) def _copy_ranges(self, files): """Copy every range of ``files`` into their temp files; return indexes that failed.""" - chunk = self._hydrate_chunk_size - work = [ - (off, i, min(chunk, f.size - off)) for i, f in enumerate(files) for off in range(0, f.size, chunk) - ] - # Offset-major order spreads concurrent readers across every file. - work.sort(key=lambda w: (w[0], files[w[1]].src)) failed = set() def run(item): @@ -375,14 +385,12 @@ def run(item): return f = files[i] try: - _with_range_retry(_copy_range, f.src, f.tmp, off, length) + _with_range_retry(_copy_range, f.src, f.tmp, off, length, f.identity) except OSError as exc: failed.add(i) log.warn(f"VolumeCache: failed to hydrate {f.src} at offset {off}: {exc}") - if work: - with ThreadPoolExecutor(max_workers=self._hydrate_workers) as pool: - list(pool.map(run, work)) + _run_bounded(_iter_ranges(files, self._hydrate_chunk_size), run, self._hydrate_workers) return failed @staticmethod @@ -653,17 +661,66 @@ def _copy_metadata(src, dst, mtime=None): def _positive_or_default(name, value, default): - # A non-positive pool size or chunk is a caller bug that best-effort mode - # would otherwise swallow mid-operation, silently disabling the cache. - if value is not None and value < 1: + # A non-positive or non-integer pool size or chunk is a caller bug that + # best-effort mode would otherwise swallow mid-operation (range() and + # ThreadPoolExecutor reject floats), silently disabling the cache. bool is + # an int subclass, so it is rejected explicitly. + if value is None: + return default + if isinstance(value, bool) or not isinstance(value, int) or value < 1: raise ValueError(f"{name} must be a positive integer, got {value!r}") - return value or default + return value def _tmp_path(path): return f"{path}.{os.getpid()}.{threading.get_ident()}.rpvc.tmp" +def _iter_ranges(files, chunk): + """Lazily yield ``(offset, index, length)`` for every range of ``files``. + + Offset-major order spreads concurrent readers across every file. Files + drop out once exhausted, so the cost is O(ranges + files) and nothing is + materialized up front. + """ + active = sorted((i for i, f in enumerate(files) if f.size > 0), key=lambda i: files[i].src) + offset = 0 + while active: + for i in active: + yield offset, i, min(chunk, files[i].size - offset) + offset += chunk + active = [i for i in active if offset < files[i].size] + + +def _run_bounded(items, fn, workers): + """Call fn on each of ``items`` across ``workers`` threads. + + Each thread pulls the next item from the shared iterator, so outstanding + work stays O(workers) however many items there are; ``pool.map`` would + submit a future per item up front. An exception escaping fn stops every + thread from taking further items and propagates once the pool shuts down. + """ + items = iter(items) + lock = threading.Lock() # generators are not thread-safe + stop = threading.Event() + + def drain(): + while not stop.is_set(): + with lock: + item = next(items, None) + if item is None: + return + try: + fn(item) + except BaseException: + stop.set() + raise + + with ThreadPoolExecutor(max_workers=workers) as pool: + for future in [pool.submit(drain) for _ in range(workers)]: + future.result() + + def _with_range_retry(fn, *args): """Call fn, retrying transient FUSE errors with exponential backoff.""" for attempt in range(_RANGE_MAX_ATTEMPTS): @@ -675,16 +732,33 @@ def _with_range_retry(fn, *args): time.sleep(_RANGE_RETRY_BASE_SECONDS * 2**attempt) -def _copy_range(src, dst, offset, length): +def _file_identity(st): + """The generation of a file: changes when sync() replaces it. + + Size and mtime only. Inode numbers on a FUSE object-store mount are + assigned by the filesystem and can change for an unchanged object (e.g. + after eviction), which would reject valid files, while a cross-host + replacement there keeps the inode and changes size or mtime instead. + """ + return (st.st_size, st.st_mtime_ns) + + +def _copy_range(src, dst, offset, length, identity): """Copy ``length`` bytes at ``offset`` of src to the same offset of dst. Each call opens its own descriptors so an object-store mount serves the - range as an independent read. A short read means the mirror file is - smaller than the manifest says; it raises without an errno, so it is not - retried. + range as an independent read. The opened source must still match the + ``identity`` pinned before the first range, so a replacement by a + concurrent sync() seen between ranges fails the file instead of splicing + two generations together. An object-store mount resolves reads by key, so + a replacement landing mid-read is detected only by later ranges. A + short read means the mirror file is smaller than the manifest says. Both + raise without an errno, so they are not retried. """ src_fd = os.open(src, os.O_RDONLY) try: + if _file_identity(os.fstat(src_fd)) != identity: + raise OSError(f"source replaced during hydrate at offset {offset}") dst_fd = os.open(dst, os.O_WRONLY) try: done = 0 diff --git a/tests/test_serverless/test_utils/test_rp_volume_cache.py b/tests/test_serverless/test_utils/test_rp_volume_cache.py index 492dce118..a52648e78 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -1146,7 +1146,7 @@ def test_hydrate_copies_big_file_in_ranges(tmp_path, monkeypatch): ranges = [] real_copy_range = vcmod._copy_range monkeypatch.setattr( - vcmod, "_copy_range", lambda *a: ranges.append(a[2:]) or real_copy_range(*a) + vcmod, "_copy_range", lambda *a: ranges.append(a[2:4]) or real_copy_range(*a) ) restored = _hydrating_cache(cache, vol).hydrate() @@ -1239,6 +1239,98 @@ def test_hydrate_tuning_rejects_non_positive(tmp_path, arg, bad): VolumeCache([str(tmp_path / "cache")], namespace="ep1", volume_path=str(vol), **{arg: bad}) +@pytest.mark.parametrize("arg", ["max_workers", "hydrate_workers", "hydrate_chunk_size"]) +@pytest.mark.parametrize("bad", [8388608.0, 2.5, True, "64"]) +def test_tuning_rejects_non_integer(tmp_path, arg, bad): + # A float chunk passes a `< 1` check but fails in range() mid-hydrate, after + # temp files exist; reject it at construction instead. + vol = tmp_path / "volume" + vol.mkdir() + with pytest.raises(ValueError): + VolumeCache([str(tmp_path / "cache")], namespace="ep1", volume_path=str(vol), **{arg: bad}) + + +def test_hydrate_rejects_source_replaced_between_ranges(tmp_path, monkeypatch): + # A concurrent sync() atomically replaces the mirror file. Ranges read from + # two generations would assemble a mixed file of the right size; hydrate must + # give that file up instead of publishing it. + cache, vol, payloads = _synced_big_files(tmp_path, monkeypatch, ["model.safetensors"]) + mirrored = vol / ".cache" / "ep1" / "big" / os.path.relpath(str(cache / "model.safetensors"), "/") + replacement = mirrored.with_name("replacement") + replacement.write_bytes(os.urandom(len(payloads["model.safetensors"]))) + real_copy_range = vcmod._copy_range + calls = [] + + def copy_then_replace(*args): + real_copy_range(*args) + calls.append(args) + if len(calls) == 1: + os.replace(replacement, mirrored) + + monkeypatch.setattr(vcmod, "_copy_range", copy_then_replace) + + restored = _hydrating_cache(cache, vol, hydrate_workers=1).hydrate() + + assert restored == 0 + assert not (cache / "model.safetensors").exists() + assert [p for p in os.listdir(cache) if p.endswith(".rpvc.tmp")] == [] + + +def test_hydrate_rejects_source_whose_size_differs_from_manifest(tmp_path, monkeypatch): + # A sync that replaced the mirror file after hydrate read the manifest would + # otherwise have only the manifest's (older, shorter) length copied out of + # the newer file, publishing a truncated model with no error. + cache, vol, payloads = _synced_big_files(tmp_path, monkeypatch, ["model.safetensors"]) + mirrored = vol / ".cache" / "ep1" / "big" / os.path.relpath(str(cache / "model.safetensors"), "/") + mirrored.write_bytes(os.urandom(len(payloads["model.safetensors"]) + _CHUNK)) + + restored = _hydrating_cache(cache, vol).hydrate() + + assert restored == 0 + assert not (cache / "model.safetensors").exists() + assert [p for p in os.listdir(cache) if p.endswith(".rpvc.tmp")] == [] + + +def test_hydrate_unexpected_error_stops_ranges_and_cleans_temps(tmp_path, monkeypatch): + # A non-OSError escaping a range (e.g. MemoryError) aborts the hydrate; the + # remaining ranges must not keep running, and no full-size temp may linger. + cache, vol, _payloads = _synced_big_files(tmp_path, monkeypatch, ["a.bin", "b.bin"]) + calls = [] + + def boom(*args): + calls.append(args) + raise RuntimeError("boom") + + monkeypatch.setattr(vcmod, "_copy_range", boom) + + restored = _hydrating_cache(cache, vol, hydrate_workers=2).hydrate() + + assert restored == 0 + assert len(calls) <= 2 + assert [p for p in os.listdir(cache) if p.endswith(".rpvc.tmp")] == [] + + +def test_hydrate_bounds_outstanding_range_tasks(tmp_path, monkeypatch): + # Scheduling must stay O(workers), not O(total bytes / chunk): a large + # tree with a small chunk would otherwise queue millions of futures. + cache, vol, payloads = _synced_big_files(tmp_path, monkeypatch, ["a.bin", "b.bin"]) + submitted = [] + + class CountingPool(vcmod.ThreadPoolExecutor): + def submit(self, *args, **kwargs): + submitted.append(args) + return super().submit(*args, **kwargs) + + monkeypatch.setattr(vcmod, "ThreadPoolExecutor", CountingPool) + + restored = _hydrating_cache(cache, vol, hydrate_workers=2).hydrate() + + assert restored == 2 + assert len(submitted) <= 2 + for name, payload in payloads.items(): + assert (cache / name).read_bytes() == payload + + def test_hydrate_tuning_defaults(tmp_path): vc, _cache, _vol = _mk_cache_with_volume(tmp_path) assert vc._hydrate_workers == 64 From d29de27f685ce92e5fd2d3fe2ceb98d1ddf05e88 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Thu, 15 Oct 2026 17:28:15 -0700 Subject: [PATCH 3/5] refactor(serverless): make range retry's final attempt explicit Hoist the last attempt out of the retry loop so _with_range_retry has no implicit fall-through return (CodeQL py/mixed-returns). Adds a test pinning the attempt count when transient errors persist. --- runpod/serverless/utils/rp_volume_cache.py | 5 +++-- .../test_serverless/test_utils/test_rp_volume_cache.py | 10 ++++++++++ 2 files changed, 13 insertions(+), 2 deletions(-) diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index ccf6924d7..cc5cefa55 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -723,13 +723,14 @@ def drain(): def _with_range_retry(fn, *args): """Call fn, retrying transient FUSE errors with exponential backoff.""" - for attempt in range(_RANGE_MAX_ATTEMPTS): + for attempt in range(_RANGE_MAX_ATTEMPTS - 1): try: return fn(*args) except OSError as exc: - if exc.errno not in _RANGE_RETRY_ERRNOS or attempt == _RANGE_MAX_ATTEMPTS - 1: + if exc.errno not in _RANGE_RETRY_ERRNOS: raise time.sleep(_RANGE_RETRY_BASE_SECONDS * 2**attempt) + return fn(*args) # final attempt: any error propagates def _file_identity(st): diff --git a/tests/test_serverless/test_utils/test_rp_volume_cache.py b/tests/test_serverless/test_utils/test_rp_volume_cache.py index a52648e78..f24fa072c 100644 --- a/tests/test_serverless/test_utils/test_rp_volume_cache.py +++ b/tests/test_serverless/test_utils/test_rp_volume_cache.py @@ -1195,6 +1195,16 @@ def test_hydrate_does_not_retry_non_transient_errors(tmp_path, monkeypatch): assert len(failures) == 1 # first failure aborts the file; its other ranges are skipped +def test_hydrate_gives_up_after_max_range_attempts(tmp_path, monkeypatch): + cache, vol, _payloads = _synced_big_files(tmp_path, monkeypatch, ["model.safetensors"]) + failures = _fail_pread_on(monkeypatch, "/big/", errno.EIO) + + restored = _hydrating_cache(cache, vol, hydrate_workers=1).hydrate() + + assert restored == 0 + assert len(failures) == vcmod._RANGE_MAX_ATTEMPTS + + def test_hydrate_skips_only_the_failing_file(tmp_path, monkeypatch): cache, vol, payloads = _synced_big_files(tmp_path, monkeypatch, ["bad.bin", "good.bin"]) _fail_pread_on(monkeypatch, "bad.bin", errno.EIO) From d762ee80e26754ff995b112810441c10925d26c2 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Thu, 15 Oct 2026 17:41:56 -0700 Subject: [PATCH 4/5] chore(ci): silence runpod-flash sunset warning in e2e runpod-flash 1.21.0 emits a FutureWarning on first public-API access, which pytest turns into a conftest ImportError. Set RUNPOD_FLASH_SILENCE_DEPRECATION=1 for the e2e step. --- .github/workflows/ci.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index dfc262c47..d866c0f8d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -84,6 +84,8 @@ jobs: env: RUNPOD_API_KEY: ${{ secrets.RUNPOD_API_KEY }} RUNPOD_SDK_GIT_REF: ${{ github.event_name == 'pull_request' && github.event.pull_request.head.sha || github.sha }} + # runpod-flash >=1.21 emits a sunset FutureWarning on import, which pytest treats as an error + RUNPOD_FLASH_SILENCE_DEPRECATION: "1" codeql: if: github.event_name == 'schedule' || github.event_name == 'pull_request' From 92fe352184f6d99c31503bb8ab6ab3bb446d3007 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Dean=20Qui=C3=B1anola?= Date: Thu, 15 Oct 2026 17:45:41 -0700 Subject: [PATCH 5/5] refactor(serverless): treat range retry as a procedure _with_range_retry only wraps _copy_range, which returns None, so stop forwarding its result (CodeQL py/procedure-return-value-used). --- runpod/serverless/utils/rp_volume_cache.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/runpod/serverless/utils/rp_volume_cache.py b/runpod/serverless/utils/rp_volume_cache.py index cc5cefa55..014a430a1 100644 --- a/runpod/serverless/utils/rp_volume_cache.py +++ b/runpod/serverless/utils/rp_volume_cache.py @@ -725,12 +725,13 @@ def _with_range_retry(fn, *args): """Call fn, retrying transient FUSE errors with exponential backoff.""" for attempt in range(_RANGE_MAX_ATTEMPTS - 1): try: - return fn(*args) + fn(*args) + return except OSError as exc: if exc.errno not in _RANGE_RETRY_ERRNOS: raise time.sleep(_RANGE_RETRY_BASE_SECONDS * 2**attempt) - return fn(*args) # final attempt: any error propagates + fn(*args) # final attempt: any error propagates def _file_identity(st):