diff --git a/fastsafetensors/common.py b/fastsafetensors/common.py
index 5f74b83..e004867 100644
--- a/fastsafetensors/common.py
+++ b/fastsafetensors/common.py
@@ -6,7 +6,7 @@
 import sys
 from collections import OrderedDict
 from dataclasses import dataclass
-from typing import Callable, Dict, List, Optional, Tuple
+from typing import Callable, Dict, List, Optional, Set, Tuple
 
 from . import cpp as fstcpp
 from .dlpack import from_cuda_buffer
@@ -37,6 +37,31 @@ def is_gpu_found():
     return fstcpp.is_cuda_found() or fstcpp.is_hip_found()
 
 
+def get_fs_type(path: str, mounts_file: str = "/proc/mounts") -> str:
+    """Best-effort filesystem type for *path* (longest mount-point prefix
+    match). Returns "" when it cannot be determined (non-Linux, unreadable
+    mounts table). Used to pick I/O strategies -- e.g. O_DIRECT is a win on
+    local block devices but loses kernel readahead on network filesystems.
+    """
+    try:
+        real = os.path.realpath(path)
+        best_mnt, best_type = "", ""
+        with open(mounts_file) as f:
+            for line in f:
+                parts = line.split()
+                if len(parts) < 3:
+                    continue
+                # /proc/mounts octal-escapes spaces and tabs in mount points
+                mnt = parts[1].replace("\\040", " ").replace("\\011", "\t")
+                fstype = parts[2]
+                if real == mnt or real.startswith(mnt.rstrip("/") + "/"):
+                    if len(mnt) > len(best_mnt):
+                        best_mnt, best_type = mnt, fstype
+        return best_type
+    except OSError:
+        return ""
+
+
 def get_device_numa_node(device: Optional[int]) -> Optional[int]:
     if device is None or not sys.platform.startswith("linux"):
         return None
@@ -340,9 +365,15 @@ def get_tensors(
         device: Device,
         copy_start_offset: int,
         dtype: DType = DType.AUTO,
+        names: Optional[Set[str]] = None,
     ) -> Dict[str, TensorBase]:
+        # ``names`` restricts instantiation to that subset of tensors. Required
+        # when ``gbuf`` is a compacted chunk buffer (holding only some tensors'
+        # bytes): other tensors' computed offsets would fall outside the buffer.
         ret = {}
         for tensor_name, t in self.tensors.items():
+            if names is not None and tensor_name not in names:
+                continue
             dst_dev_ptr = (
                 gbuf.get_base_address()
                 + self.header_length
@@ -428,6 +459,62 @@ def select_byte_ranges(
                 merged.append([s, e])
         return [(s, e) for s, e in merged]
 
+    def plan_chunks(
+        self,
+        max_batch_bytes: int,
+        keep_tensor: Optional[Callable[[str], bool]] = None,
+        merge_gap: int = 4096,
+    ) -> List[Tuple[Set[str], List[Tuple[int, int]]]]:
+        """Partition this shard's tensors into byte-budgeted sub-file chunks.
+
+        Walks the (optionally ``keep_tensor``-filtered) tensors in file-offset
+        order and greedily packs consecutive tensors into a chunk while the
+        chunk's byte span stays within ``max_batch_bytes``. Returns a list of
+        ``(names, byte_ranges)`` pairs; feed each to a copier's ``set_chunk`` to
+        load only that chunk, bounding peak device memory to ~``max_batch_bytes``
+        (the device buffer covers the chunk's span). The byte ranges are the
+        kept tensors' runs within the chunk (kept tensors separated by at most
+        ``merge_gap`` bytes are coalesced), so holes left by a filter are not
+        read even though the buffer spans them.
+
+        A tensor is the atomic load unit, so ``max_batch_bytes`` must be at least
+        the largest kept tensor -- otherwise that tensor fits in no chunk and a
+        ``ValueError`` is raised naming it.
+        """
+        chunks: List[Tuple[Set[str], List[Tuple[int, int]]]] = []
+        cur: List[str] = []
+        cur_runs: List[List[int]] = []  # merged kept runs within the chunk
+        cur_start = 0
+
+        def _push(name: str, s: int, e: int) -> None:
+            if cur_runs and s - cur_runs[-1][1] <= merge_gap:
+                cur_runs[-1][1] = max(cur_runs[-1][1], e)
+            else:
+                cur_runs.append([s, e])
+
+        for name, frame in self.tensors.items():  # insertion order == offset order
+            if keep_tensor is not None and not keep_tensor(name):
+                continue
+            s = self.header_length + frame.data_offsets[0]
+            e = self.header_length + frame.data_offsets[1]
+            tensor_bytes = frame.data_offsets[1] - frame.data_offsets[0]
+            if tensor_bytes > max_batch_bytes:
+                raise ValueError(
+                    f"max_batch_bytes={max_batch_bytes} is smaller than tensor "
+                    f"'{name}' ({tensor_bytes} bytes); it must be at least the "
+                    f"largest kept tensor so every tensor fits in one chunk."
+                )
+            if cur and (e - cur_start) > max_batch_bytes:
+                chunks.append((set(cur), [(s0, e0) for s0, e0 in cur_runs]))
+                cur, cur_runs = [], []
+            if not cur:
+                cur_start = s
+            cur.append(name)
+            _push(name, s, e)
+        if cur:
+            chunks.append((set(cur), [(s0, e0) for s0, e0 in cur_runs]))
+        return chunks
+
     def __repr__(self) -> str:
         return str({"__metadata__": self.metadata, "tensors": self.tensors})
 
diff --git a/fastsafetensors/config.py b/fastsafetensors/config.py
index 9b6bfb6..f55542b 100644
--- a/fastsafetensors/config.py
+++ b/fastsafetensors/config.py
@@ -3,7 +3,7 @@
 import json
 import os
 from dataclasses import dataclass, field, fields
-from typing import Any, Dict
+from typing import Any, Dict, Optional
 
 from .common import init_logger
 
@@ -33,6 +33,11 @@ class LoaderConfig:
     queue_size: int = 0
     use_tqdm_on_load: bool = True
 
+    # Cap peak device-buffer bytes per rank by loading each shard in sub-file
+    # chunks (load -> [broadcast] -> release). None keeps whole-shard loading.
+    # Must be >= the largest single tensor. See SafeTensorsMetadata.plan_chunks.
+    max_batch_bytes: Optional[int] = None
+
     _extensions: Dict[str, Dict[str, Any]] = field(default_factory=dict)
 
     def __post_init__(self):
@@ -104,14 +109,19 @@ def from_file(cls, path: str) -> "LoaderConfig":
         return cls._from_json(path)
 
     def create_parallel_kwargs(self) -> Dict[str, Any]:
+        # Memory knobs apply with or without pipelining.
+        common: Dict[str, Any] = {
+            "max_batch_bytes": self.max_batch_bytes,
+        }
         if not self.use_pipeline:
             # queue_size=-1: fully serial (copy_files → broadcast → copy_files),
             # only 1 batch in GPU memory at a time.
-            return {"queue_size": -1}
+            return {"queue_size": -1, **common}
         return {
             "max_concurrent_producers": self.max_concurrent_producers,
             "queue_size": self.queue_size,
             "use_tqdm_on_load": self.use_tqdm_on_load,
+            **common,
         }
 
 
diff --git a/fastsafetensors/copier/nogds.py b/fastsafetensors/copier/nogds.py
index 054b6fc..eca586d 100644
--- a/fastsafetensors/copier/nogds.py
+++ b/fastsafetensors/copier/nogds.py
@@ -2,7 +2,7 @@
 
 import os
 import sys
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List, Optional, Set, Tuple
 
 from .. import cpp as fstcpp
 from ..common import (
@@ -40,6 +40,8 @@ def __init__(
         self.device = device
         self.reqs: List[int] = []
         self.byte_ranges: Optional[List[Tuple[int, int]]] = None
+        self._chunk_names: Optional[Set[str]] = None
+        self._base_off = metadata.header_length
 
     def set_byte_ranges(self, byte_ranges: Optional[List[Tuple[int, int]]]) -> None:
         """Restrict reads to these ``[start, end)`` absolute file-offset runs.
@@ -51,26 +53,44 @@ def set_byte_ranges(self, byte_ranges: Optional[List[Tuple[int, int]]]) -> None:
         """
         self.byte_ranges = validated_byte_ranges(self.metadata, byte_ranges)
 
+    def set_chunk(self, byte_ranges: List[Tuple[int, int]], names: Set[str]) -> None:
+        """Load only ``names`` into a device buffer sized to those tensors' span.
+
+        Unlike ``set_byte_ranges`` (which still allocates the whole data section
+        and leaves it sparsely filled), this allocates only
+        ``max_end - min_start`` of the runs and instantiates just ``names`` --
+        so peak device memory is bounded by the chunk, not the shard. This is
+        the building block for ``max_batch_bytes`` sub-file batching.
+        """
+        self.byte_ranges = byte_ranges
+        self._chunk_names = names
+
     def submit_io(
         self, use_buf_register: bool, max_copy_block_size: int
     ) -> fstcpp.gds_device_buffer:
         header_length = self.metadata.header_length
-        total_length = self.metadata.size_bytes - header_length
-        gbuf = self.framework.alloc_tensor_memory(total_length, self.device)
         # Default to a single run spanning the whole data section, which
         # reproduces the original full-file read.
         runs = self.byte_ranges
         if runs is None:
             runs = [(header_length, self.metadata.size_bytes)]
+        if self._chunk_names is not None:
+            # Compact chunk: allocate only the runs' span and map gbuf[0] to the
+            # first run's start, so peak memory tracks the chunk, not the shard.
+            base_off = min(s for s, _ in runs)
+            alloc_length = max(e for _, e in runs) - base_off
+        else:
+            base_off = header_length
+            alloc_length = self.metadata.size_bytes - header_length
+        self._base_off = base_off
+        gbuf = self.framework.alloc_tensor_memory(alloc_length, self.device)
         for start, end in runs:
             count = start
             while count < end:
                 l = end - count
                 if max_copy_block_size < l:
                     l = max_copy_block_size
-                req = self.reader.submit_read(
-                    self.fd, gbuf, count, l, count - header_length
-                )
+                req = self.reader.submit_read(self.fd, gbuf, count, l, count - base_off)
                 if req < 0:
                     raise Exception(f"submit_io: submit_nogds_read failed, err={req}")
                 self.reqs.append(req)
@@ -96,7 +116,7 @@ def wait_io(
         if len(failed) > 0:
             raise Exception(f"wait_io: wait_nogds_read failed, reqs={failed}")
         return self.metadata.get_tensors(
-            gbuf, self.device, self.metadata.header_length, dtype=dtype
+            gbuf, self.device, self._base_off, dtype=dtype, names=self._chunk_names
         )
 
 
diff --git a/fastsafetensors/copier/unified.py b/fastsafetensors/copier/unified.py
index b361143..9810334 100644
--- a/fastsafetensors/copier/unified.py
+++ b/fastsafetensors/copier/unified.py
@@ -13,15 +13,56 @@
 """
 
 import os
-from typing import Dict, List, Optional, Tuple
+from typing import Dict, List, Optional, Set, Tuple
 
 from .. import cpp as fstcpp
-from ..common import SafeTensorsMetadata
+from ..common import SafeTensorsMetadata, get_fs_type, init_logger
 from ..frameworks import FrameworkOpBase, TensorBase
 from ..st_types import Device, DType
 from .base import CopierInterface, validated_byte_ranges
 from .registry import CopierConstructFunc, register_copier_constructor
 
+logger = init_logger(__name__)
+
+# O_DIRECT bypasses the page cache, which wins on local block devices but
+# forfeits kernel readahead / client caching on network filesystems, where the
+# buffered mmap + pin path performs better. Gate the fast path by fs type;
+# FASTSAFETENSORS_ODIRECT=1/0 forces it on/off regardless.
+_NETWORK_FS = {
+    "nfs",
+    "nfs4",
+    "cifs",
+    "smb3",
+    "smbfs",
+    "sshfs",
+    "fuse.sshfs",
+    "lustre",
+    "gpfs",
+    "beegfs",
+    "glusterfs",
+    "ceph",
+    "9p",
+    "virtiofs",
+}
+_warned_fs: set = set()
+
+
+def _odirect_ok(path: str) -> bool:
+    override = os.environ.get("FASTSAFETENSORS_ODIRECT")
+    if override is not None:
+        return override == "1"
+    fstype = get_fs_type(path)
+    if fstype in _NETWORK_FS:
+        if fstype not in _warned_fs:
+            _warned_fs.add(fstype)
+            logger.info(
+                "checkpoint on network filesystem (%s): using buffered reads "
+                "instead of O_DIRECT (set FASTSAFETENSORS_ODIRECT=1 to force)",
+                fstype,
+            )
+        return False
+    return True
+
 
 class UnifiedMemCopier(CopierInterface):
     """Copier using mmap → pin_memory → cudaMemcpyAsync for unified memory.
@@ -43,6 +84,13 @@ def __init__(
         self.framework = framework
         self._pinned: List[TensorBase] = []
         self.byte_ranges: Optional[List[Tuple[int, int]]] = None
+        self._chunk_names: Optional[Set[str]] = None
+        self._base_off = metadata.header_length
+        # Worker count for the C++ O_DIRECT range reader (dma_load_runs). Single
+        # -thread pin_memory is page-cache-bound (~2.5 GB/s); O_DIRECT threads
+        # bypass the cache and drive NVMe queue depth. Falls back to pin_memory
+        # if the reader is unavailable.
+        self._dma_threads = int(os.environ.get("FASTSAFETENSORS_DMA_THREADS", "8"))
 
     def set_byte_ranges(self, byte_ranges: Optional[List[Tuple[int, int]]]) -> None:
         """Restrict reads to these ``[start, end)`` absolute file-offset runs.
@@ -55,14 +103,20 @@ def set_byte_ranges(self, byte_ranges: Optional[List[Tuple[int, int]]]) -> None:
         """
         self.byte_ranges = validated_byte_ranges(self.metadata, byte_ranges)
 
+    def set_chunk(self, byte_ranges: List[Tuple[int, int]], names: Set[str]) -> None:
+        """Load only ``names`` into a buffer sized to those tensors' span.
+
+        Allocates only ``max_end - min_start`` of the runs and instantiates just
+        ``names``, so peak device memory is bounded by the chunk rather than the
+        shard -- the building block for ``max_batch_bytes`` sub-file batching.
+        """
+        self.byte_ranges = byte_ranges
+        self._chunk_names = names
+
     def submit_io(
         self, use_buf_register: bool, max_copy_block_size: int
     ) -> fstcpp.gds_device_buffer:
         header_length = self.metadata.header_length
-        data_length = self.metadata.size_bytes - header_length
-
-        # Allocate CUDA buffer via framework's allocator (proper lifecycle)
-        gbuf = self.framework.alloc_tensor_memory(data_length, self.device)
 
         # Default to the whole data section, reproducing the full-file read.
         # An empty list (vs None) reads nothing — same semantics as nogds.
@@ -70,18 +124,56 @@ def submit_io(
         if runs is None:
             runs = [(header_length, self.metadata.size_bytes)]
 
+        if self._chunk_names is not None:
+            # Compact chunk: allocate only the runs' span and map gbuf[0] to the
+            # first run's start, so peak memory tracks the chunk, not the shard.
+            base_off = min(s for s, _ in runs)
+            alloc_length = max(e for _, e in runs) - base_off
+        else:
+            base_off = header_length
+            alloc_length = self.metadata.size_bytes - header_length
+        self._base_off = base_off
+
+        # Allocate CUDA buffer via framework's allocator (proper lifecycle)
+        gbuf = self.framework.alloc_tensor_memory(alloc_length, self.device)
+
+        # Fast path: multithreaded O_DIRECT reader copies only the runs straight
+        # into gbuf (byte F -> gbuf[F - base_off]), bypassing the page cache and
+        # single-thread pin. Works for both full and compact-chunk buffers.
+        # FASTSAFETENSORS_DMA_THREADS=0 disables it (falls back to mmap + pin);
+        # network filesystems fall back automatically (see _odirect_ok).
+        dma_load_runs = getattr(fstcpp, "dma_load_runs", None)
+        if (
+            dma_load_runs is not None
+            and self._dma_threads > 0
+            and _odirect_ok(self.metadata.src)
+        ):
+            starts = [s for s, _ in runs]
+            ends = [e for _, e in runs]
+            rc = dma_load_runs(
+                gbuf.get_base_address(),
+                self.metadata.src,
+                base_off,
+                starts,
+                ends,
+                self._dma_threads,
+            )
+            if rc == 0:
+                return gbuf
+            # Non-zero: reader unusable here; fall back to mmap + pin_memory.
+
         base_address = gbuf.get_base_address()
         self._pinned = []
         for start, end in runs:
             # mmap_file_pinned faults in + pins only this run's pages
             # (kernel readahead + DMA-ready), then DMA to the matching offset in
-            # gbuf (data section starts at header_length).
+            # gbuf (gbuf[0] maps to file offset base_off).
             pinned = self.framework.mmap_file_pinned(
                 self.metadata.src, end - start, start
             )
             self._pinned.append(pinned)
             ret = fstcpp.memcpy_h2d_async(  # type: ignore[attr-defined]
-                base_address + (start - header_length),
+                base_address + (start - base_off),
                 pinned.data_ptr(),
                 end - start,
             )
@@ -107,7 +199,7 @@ def wait_io(
         # address. The copy_start_offset=header_length cancels out in get_tensors'
         # pointer arithmetic, giving correct offsets. No memmove fixup needed.
         tensors = self.metadata.get_tensors(
-            gbuf, self.device, self.metadata.header_length, dtype=dtype
+            gbuf, self.device, self._base_off, dtype=dtype, names=self._chunk_names
         )
 
         # Release the pinned mmap pages
diff --git a/fastsafetensors/cpp/ext.cpp b/fastsafetensors/cpp/ext.cpp
index 3c91876..cdfc4af 100644
--- a/fastsafetensors/cpp/ext.cpp
+++ b/fastsafetensors/cpp/ext.cpp
@@ -108,6 +108,10 @@ static inline int munmap(void* addr, size_t /*length*/) {
 #include <chrono>
 #include <cstdlib>
 #include <algorithm>
+#include <atomic>
+#include <thread>
+#include <vector>
+#include <mutex>
 
 #include "gpu_compat.h"
 #include "ext.hpp"
@@ -972,6 +976,119 @@ cpp_metrics_t get_cpp_metrics() {
 
 // Bindings
 
+// Multithreaded O_DIRECT range reader for the unified copier: reads only the
+// [starts[i], ends[i]) file runs into a device buffer, placing file byte F at
+// gbuf[F - header_len]. Threads split the concatenated owned-byte space so one
+// large run + many small runs still spread evenly; thread boundaries mid-run
+// align the O_DIRECT offset down and overlapping reads write identical bytes
+// (idempotent). Bypasses the page cache (O_DIRECT) and drives NVMe queue depth,
+// which buffered mmap+pin cannot. Uses the dlopen'd cuda_fns table (no cudart
+// link) -- pinned bounce + sync cudaMemcpy. header_len is the buffer-base offset,
+// so a compacted chunk buffer passes its span start instead of the real header.
+// Reusable pinned 16MB bounce buffers, shared across dma_load_runs calls (and
+// concurrent producers). Chunked loading makes many small calls; recycling the
+// pinned buffers avoids a cudaHostAlloc/cudaFreeHost per chunk per thread.
+static std::mutex g_pin_mtx;
+static std::vector<void *> g_pin_pool;
+static const size_t PIN_CHUNK = 16UL << 20;
+
+static void *pin_acquire() {
+    {
+        std::lock_guard<std::mutex> lk(g_pin_mtx);
+        if (!g_pin_pool.empty()) {
+            void *p = g_pin_pool.back();
+            g_pin_pool.pop_back();
+            return p;
+        }
+    }
+    void *p = nullptr;
+    if (cuda_fns.cudaHostAlloc(&p, PIN_CHUNK, 0) != cudaSuccess) return nullptr;
+    return p;
+}
+
+static void pin_release(void *p) {
+    if (!p) return;
+    std::lock_guard<std::mutex> lk(g_pin_mtx);
+    g_pin_pool.push_back(p);
+}
+
+static int dma_load_runs(uintptr_t gbuf_dev, const std::string &path,
+                         size_t header_len,
+                         const std::vector<size_t> &starts,
+                         const std::vector<size_t> &ends, int nthreads) {
+    if (!cuda_fns.cudaHostAlloc || !cuda_fns.cudaMemcpy || !cuda_fns.cudaFreeHost
+        || !cuda_fns.cudaDeviceSynchronize) {
+        return -10;
+    }
+    const size_t n_runs = starts.size();
+    if (n_runs == 0 || ends.size() != n_runs) return 0;
+    size_t total = 0;
+    for (size_t i = 0; i < n_runs; i++) total += ends[i] - starts[i];
+    if (total == 0) return 0;
+    if (nthreads < 1) nthreads = 4;
+    if (nthreads > 32) nthreads = 32;
+
+    char *gbuf = reinterpret_cast<char *>(gbuf_dev);
+    const size_t CHUNK = 16UL << 20;
+    const size_t ALN = 4096UL;
+    std::atomic<int> rc{0};
+    std::vector<std::thread> threads;
+
+    for (int ti = 0; ti < nthreads; ti++) {
+        size_t gbs = (size_t)((double)ti * total / nthreads);
+        size_t gbe = (ti == nthreads - 1) ? total
+                                          : (size_t)((double)(ti + 1) * total / nthreads);
+        if (gbe <= gbs) continue;
+        threads.emplace_back([&, gbs, gbe]() {
+            void *pinned = pin_acquire();
+            if (!pinned) {
+                rc = -1;
+                return;
+            }
+            int fd = open(path.c_str(), O_RDONLY | O_DIRECT);
+            if (fd < 0) {
+                rc = -2;
+                pin_release(pinned);
+                return;
+            }
+            size_t cum = 0;
+            for (size_t r = 0; r < n_runs && rc.load() == 0; r++) {
+                size_t rs = starts[r], re = ends[r], rlen = re - rs;
+                size_t b0 = cum, b1 = cum + rlen;  // this run in owned-byte space
+                cum = b1;
+                size_t ov0 = b0 > gbs ? b0 : gbs;  // overlap with my span
+                size_t ov1 = b1 < gbe ? b1 : gbe;
+                if (ov0 >= ov1) continue;
+                size_t fstart = rs + (ov0 - b0);   // file coords of my portion
+                size_t fend = rs + (ov1 - b0);
+                size_t astart = fstart & ~(ALN - 1);  // align O_DIRECT offset down
+                for (size_t fo = astart; fo < fend; fo += CHUNK) {
+                    size_t want = fend - fo;
+                    size_t reqlen = (want >= CHUNK) ? CHUNK
+                                                    : ((want + ALN - 1) & ~(ALN - 1));
+                    ssize_t got = pread(fd, pinned, reqlen, fo);
+                    if (got <= 0) { rc = -3; break; }
+                    size_t fo_end = fo + (size_t)got;
+                    size_t cs = fo > fstart ? fo : fstart;  // copy only [fstart,fend)
+                    size_t ce = fo_end < fend ? fo_end : fend;
+                    if (cs < ce) {
+                        cudaError_t e = cuda_fns.cudaMemcpy(
+                            gbuf + (cs - header_len), (char *)pinned + (cs - fo),
+                            ce - cs, cudaMemcpyHostToDevice);
+                        if (e != cudaSuccess) { rc = -4; break; }
+                    }
+                    if (fo_end >= fend) break;
+                }
+            }
+            close(fd);
+            pin_release(pinned);
+        });
+    }
+    for (auto &t : threads) t.join();
+    cuda_fns.cudaDeviceSynchronize();
+    return rc.load();
+}
+
 // Async host-to-device memcpy for unified memory copier
 static int memcpy_h2d_async(uintptr_t dst, uintptr_t src, size_t size) {
     if (!cuda_fns.cudaMemcpyAsync) {
@@ -1013,6 +1130,17 @@ PYBIND11_MODULE(__MOD_NAME__, m)
     m.def("load_library_functions", &load_library_functions,
           pybind11::arg("cudart_lib_name") = "");
     m.def("memcpy_h2d_async", &memcpy_h2d_async);
+    m.def(
+        "dma_load_runs",
+        [](uintptr_t gbuf_dev, const std::string &path, size_t header_len,
+           const std::vector<size_t> &starts, const std::vector<size_t> &ends,
+           int nthreads) {
+            pybind11::gil_scoped_release release;  // blocking O_DIRECT + DMA
+            return dma_load_runs(gbuf_dev, path, header_len, starts, ends, nthreads);
+        },
+        pybind11::arg("gbuf_dev"), pybind11::arg("path"),
+        pybind11::arg("header_len"), pybind11::arg("starts"),
+        pybind11::arg("ends"), pybind11::arg("nthreads") = 8);
     m.def("get_cpp_metrics", &get_cpp_metrics);
     m.def("set_gil_release", &set_gil_release);
     m.def("get_gil_release", &get_gil_release);
diff --git a/fastsafetensors/loader.py b/fastsafetensors/loader.py
index 466f710..2c00a46 100644
--- a/fastsafetensors/loader.py
+++ b/fastsafetensors/loader.py
@@ -10,6 +10,7 @@
     Mapping,
     Optional,
     OrderedDict,
+    Set,
     Tuple,
     Union,
 )
@@ -75,6 +76,10 @@ def __init__(
         self.frames = OrderedDict[str, TensorFrame]()
         self.disable_cache = disable_cache
         self._tensor_filter: Optional[Callable[[str], bool]] = None
+        # realpath -> (chunk tensor names, byte-ranges) for the next
+        # copy_files_to_device; set by PipelineParallel when max_batch_bytes is
+        # active so each file loads only a sub-file chunk. Empty = whole files.
+        self._chunk_plan: Dict[str, Tuple[Set[str], List[Tuple[int, int]]]] = {}
         self.init_numa(set_numa)
         self.copier_constructor: CopierConstructFunc = create_copier_constructor(
             copier_type=copier_type,
@@ -91,9 +96,19 @@ def init_numa(self, set_numa: bool = True):
                 fstcpp.set_numa_node(node)
             gl_set_numa = True
 
+    def set_chunk_plan(
+        self, chunk_plan: Dict[str, Tuple[Set[str], List[Tuple[int, int]]]]
+    ) -> None:
+        """Load only a sub-file chunk of each listed file on the next
+        copy_files_to_device: ``realpath -> (tensor names, byte-ranges)``. The
+        copier allocates just the chunk's span (see ``set_chunk``) and only the
+        named tensors are registered. Used by max_batch_bytes batching."""
+        self._chunk_plan = chunk_plan
+
     def reset(self):
         self.frames = {}
         self.meta = {}
+        self._chunk_plan = {}
 
     def close(self):
         self.reset()
@@ -171,11 +186,26 @@ def copy_files_to_device(
 
         factory_idx_bits = math.ceil(math.log2(len(self.meta) + 1))
         lidx = 1
-        for _, (meta, rank) in sorted(self.meta.items(), key=lambda x: x[0]):
+        for realpath, (meta, rank) in sorted(self.meta.items(), key=lambda x: x[0]):
             self_rank = self.pg.rank() == rank
             if self_rank:
                 copier = self.copier_constructor(meta, self.device, self.framework)
-                if self._tensor_filter is not None:
+                chunk = self._chunk_plan.get(realpath)
+                if chunk is not None:
+                    if not hasattr(copier, "set_chunk"):
+                        # Silently loading the whole file per chunk-batch would
+                        # both break the memory bound and multiply full-file
+                        # reads -- refuse instead.
+                        raise NotImplementedError(
+                            f"sub-file chunking (max_batch_bytes / "
+                            f"device_memory_budget) requires a copier with "
+                            f"set_chunk; {type(copier).__name__} loads whole "
+                            f"files. Use the nogds or unified copier, or unset "
+                            f"the chunking options."
+                        )
+                    names, ranges = chunk
+                    copier.set_chunk(ranges, names)
+                elif self._tensor_filter is not None:
                     copier.set_byte_ranges(meta.select_byte_ranges(self._tensor_filter))
             else:
                 copier = None
@@ -197,11 +227,17 @@ def copy_files_to_device(
             lidx += 1
         for factory in need_wait:
             factory.wait_io(dtype=dtype, noalign=False)
+        if self._chunk_plan:
+            # Only this sub-batch's chunk tensors should be registered/visible.
+            chunk_keys = set().union(*(names for names, _ in self._chunk_plan.values()))
+            keep_tensor: Optional[Callable[[str], bool]] = lambda n: n in chunk_keys
+        else:
+            keep_tensor = self._tensor_filter
         return FilesBufferOnDevice(
             factories,
             pg=self.pg,
             framework=self.framework,
-            keep_tensor=self._tensor_filter,
+            keep_tensor=keep_tensor,
         )
 
 
diff --git a/fastsafetensors/parallel_loader.py b/fastsafetensors/parallel_loader.py
index ad75b31..aec90e0 100644
--- a/fastsafetensors/parallel_loader.py
+++ b/fastsafetensors/parallel_loader.py
@@ -4,7 +4,7 @@
 import queue
 import threading
 import time
-from typing import Any, Callable, Generator, List, Optional, Tuple, Union
+from typing import Any, Callable, Dict, Generator, List, Optional, Set, Tuple, Union
 
 try:
     from tqdm.auto import tqdm
@@ -18,7 +18,7 @@ def tqdm(iterable, *args, **kwargs):
 
 
 from . import cpp as fstcpp
-from .common import SingleGroup
+from .common import SafeTensorsMetadata, SingleGroup
 from .frameworks import FrameworkOpBase
 from .loader import BaseSafeTensorsFileLoader, SafeTensorsFileLoader
 
@@ -139,6 +139,7 @@ def __init__(
         queue_size: int = 0,
         use_tqdm_on_load: bool = True,
         tensor_filter: Optional[Callable[[str], bool]] = None,
+        max_batch_bytes: Optional[int] = None,
         **kwargs,
     ):
 
@@ -168,8 +169,12 @@ def __init__(
         self.max_concurrent_producers = max_concurrent_producers
         self.queue_size = queue_size
         self.use_tqdm_on_load = use_tqdm_on_load
+        # When set, each shard is loaded in sub-file chunks (span <= this many
+        # bytes) so peak device buffer per rank is bounded regardless of shard
+        # size. See SafeTensorsMetadata.plan_chunks / CopierInterface.set_chunk.
+        self.max_batch_bytes = max_batch_bytes
 
-        # Batch files
+        # Batch files (or, with max_batch_bytes, sub-file chunk-batches)
         self.weight_files_batches = self._create_batches(pg)
 
         # Producer-consumer communication
@@ -198,7 +203,7 @@ def __init__(
 
         fstcpp.set_gil_release(True)
 
-    def _create_batches(self, pg) -> List[List[str]]:
+    def _create_batches(self, pg) -> List[List[Any]]:
         """Create file batches based on distributed settings.
 
         In distributed mode, files are grouped by the process group size so that
@@ -212,10 +217,43 @@ def _create_batches(self, pg) -> List[List[str]]:
                             files for all processes in the group
         """
         batch_size = pg.size()
-        return [
+        file_batches = [
             self.hf_weights_files[i : i + batch_size]
             for i in range(0, len(self.hf_weights_files), batch_size)
         ]
+        if self.max_batch_bytes is None:
+            return file_batches
+        # Sub-file chunking: expand each file-batch (one file per rank) into
+        # aligned chunk-batches. Chunk-batch j holds rank r's j-th chunk (or
+        # None once that rank's file runs out), so every rank issues the same
+        # broadcast sequence in lockstep. Header reads are deterministic, so all
+        # ranks build identical batches. Each shard stays owned by one rank and
+        # is loaded in chunks over successive batches -> peak buffer bounded by
+        # max_batch_bytes per rank.
+        keep = self.loader._tensor_filter
+        fw = self.loader.framework
+        chunk_batches: List[List[Any]] = []
+        for group in file_batches:
+            planned = [
+                (
+                    f,
+                    SafeTensorsMetadata.from_file(f, fw).plan_chunks(
+                        self.max_batch_bytes, keep_tensor=keep
+                    ),
+                )
+                for f in group
+            ]
+            maxn = max((len(chunks) for _, chunks in planned), default=0)
+            for j in range(maxn):
+                spec: List[Any] = []
+                for f, chunks in planned:
+                    if j < len(chunks):
+                        names, ranges = chunks[j]
+                        spec.append((f, names, ranges))
+                    else:
+                        spec.append(None)
+                chunk_batches.append(spec)
+        return chunk_batches
 
     def _log_message(self, message: str, is_error: bool = False):
         """Unified logging method for conditional print statements.
@@ -229,7 +267,26 @@ def _log_message(self, message: str, is_error: bool = False):
     def _log_error(self, message: str):
         self._log_message(message, is_error=True)
 
-    def _load_single_batch(self, batch_id: int, file_list: List[str]):
+    def _spec_to_maps(self, spec: List[Any]):
+        """Turn a batch spec into (rank_file_map, chunk_plan).
+
+        Without max_batch_bytes the spec is a list of files (one per rank). With
+        it, the spec is a chunk-batch: per rank, either (file, names, ranges) or
+        None (that rank has no chunk this batch).
+        """
+        if self.max_batch_bytes is None:
+            return {i: [f] for i, f in enumerate(spec)}, None
+        rank_file_map: Dict[int, List[str]] = {}
+        chunk_plan: Dict[str, Tuple[Set[str], List[Tuple[int, int]]]] = {}
+        for r, entry in enumerate(spec):
+            if entry is None:
+                continue
+            f, names, ranges = entry
+            rank_file_map[r] = [f]
+            chunk_plan[f] = (names, ranges)
+        return rank_file_map, chunk_plan
+
+    def _load_single_batch(self, batch_id: int, file_list: List[Any]):
         """Load a single batch into device memory.
 
         This method handles the complete process of loading a batch of files:
@@ -253,11 +310,12 @@ def _load_single_batch(self, batch_id: int, file_list: List[str]):
             return
 
         try:
-            # Prepare file mapping
-            rank_file_map = {i: [f] for i, f in enumerate(file_list)}
+            rank_file_map, chunk_plan = self._spec_to_maps(file_list)
 
             with TimingContext("add_filenames", self._log_message, batch_id) as timer:
                 self.loader.add_filenames(rank_file_map)
+                if chunk_plan is not None:
+                    self.loader.set_chunk_plan(chunk_plan)
             add_filenames_time = timer.elapsed_ms
 
             # For unbuffered behavior, wait for consumer to process previous item
@@ -492,6 +550,7 @@ def __init__(
         framework="pytorch",
         tensor_filter: Optional[Callable[[str], bool]] = None,
         all_local: bool = False,
+        max_batch_bytes: Optional[int] = None,
         **kwargs,
     ):
         """Initialize PipelineParallelLoader with a pre-configured SafeTensorsFileLoader.
@@ -535,5 +594,6 @@ def __init__(
             queue_size,
             use_tqdm_on_load,
             tensor_filter=tensor_filter,
+            max_batch_bytes=max_batch_bytes,
             **kwargs,
         )
diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py
index 6bee254..6190859 100644
--- a/tests/unit/test_config.py
+++ b/tests/unit/test_config.py
@@ -364,6 +364,7 @@ def test_create_parallel_kwargs_pipeline_enabled(self):
             "max_concurrent_producers": 1,
             "queue_size": 2,
             "use_tqdm_on_load": False,
+            "max_batch_bytes": None,
         }
 
     def test_create_parallel_kwargs_pipeline_disabled(self):
@@ -374,7 +375,10 @@ def test_create_parallel_kwargs_pipeline_disabled(self):
             use_tqdm_on_load=False,
         )
         kwargs = config.create_parallel_kwargs()
-        assert kwargs == {"queue_size": -1}
+        assert kwargs == {
+            "queue_size": -1,
+            "max_batch_bytes": None,
+        }
 
     def test_max_concurrent_producers_validation(self):
         """max_concurrent_producers != 1 should raise ValueError."""
diff --git a/tests/unit/test_fastsafetensors.py b/tests/unit/test_fastsafetensors.py
index 1ef2fbc..3b0ae03 100644
--- a/tests/unit/test_fastsafetensors.py
+++ b/tests/unit/test_fastsafetensors.py
@@ -400,6 +400,9 @@ def fake_memcpy_h2d_async(dst, src, size):
         monkeypatch.setattr(fstcpp, "memcpy_h2d_async", fake_memcpy_h2d_async)
         monkeypatch.setattr(torch.cuda, "synchronize", lambda: None)
 
+    # This test exercises the mmap + pin_memory flow; disable the O_DIRECT
+    # fast path so the monkeypatched primitives are actually used.
+    monkeypatch.setenv("FASTSAFETENSORS_DMA_THREADS", "0")
     copier = UnifiedMemCopier(meta, device, framework)
     gbuf = copier.submit_io(False, 10 * 1024 * 1024 * 1024)
     tensors = copier.wait_io(gbuf)
@@ -426,6 +429,9 @@ def test_UnifiedMemCopier_cuda_error(
     monkeypatch.setattr(torch.Tensor, "pin_memory", lambda self: self)
     monkeypatch.setattr(fstcpp, "memcpy_h2d_async", lambda dst, src, size: 99)
 
+    # Error injection targets the pin path; the O_DIRECT fast path would
+    # succeed and never hit the monkeypatched memcpy.
+    monkeypatch.setenv("FASTSAFETENSORS_DMA_THREADS", "0")
     copier = UnifiedMemCopier(meta, device, framework)
     with pytest.raises(RuntimeError, match="99"):
         copier.submit_io(False, 10 * 1024 * 1024 * 1024)
diff --git a/tests/unit/test_robustness.py b/tests/unit/test_robustness.py
index 76572ce..d313cde 100644
--- a/tests/unit/test_robustness.py
+++ b/tests/unit/test_robustness.py
@@ -1,8 +1,97 @@
 # SPDX-License-Identifier: Apache-2.0
-"""Robustness tests: graceful degradation of I/O paths."""
+"""Robustness tests: filesystem-aware I/O paths and graceful degradation."""
 
 import pytest
 
+from fastsafetensors.common import get_fs_type
+
+# ---- get_fs_type: longest-prefix mount matching ----
+
+MOUNTS = """\
+/dev/root / ext4 rw 0 0
+nas:/vol /mnt/nfs nfs4 rw 0 0
+/dev/nvme0n1p2 /data ext4 rw 0 0
+nas:/models /data/remote nfs rw 0 0
+tmpfs /dev/shm tmpfs rw 0 0
+/dev/sdb1 /mnt/with\\040space ext4 rw 0 0
+"""
+
+
+@pytest.fixture()
+def mounts_file(tmp_path):
+    p = tmp_path / "mounts"
+    p.write_text(MOUNTS)
+    return str(p)
+
+
+def test_get_fs_type_basic(mounts_file):
+    assert get_fs_type("/mnt/nfs/model.safetensors", mounts_file) == "nfs4"
+    assert get_fs_type("/data/m/x.safetensors", mounts_file) == "ext4"
+    assert get_fs_type("/somewhere/else", mounts_file) == "ext4"  # root fallback
+
+
+def test_get_fs_type_longest_prefix_wins(mounts_file):
+    # /data is ext4 but /data/remote is an NFS mount inside it
+    assert get_fs_type("/data/remote/model.safetensors", mounts_file) == "nfs"
+
+
+def test_get_fs_type_escaped_mountpoint(mounts_file):
+    assert get_fs_type("/mnt/with space/f.safetensors", mounts_file) == "ext4"
+
+
+def test_get_fs_type_unreadable_mounts(tmp_path):
+    assert get_fs_type("/data/x", str(tmp_path / "nope")) == ""
+
+
+# ---- O_DIRECT gating on network filesystems ----
+
+
+def test_odirect_gating(monkeypatch):
+    from fastsafetensors.copier import unified
+
+    monkeypatch.delenv("FASTSAFETENSORS_ODIRECT", raising=False)
+    monkeypatch.setattr(unified, "get_fs_type", lambda p: "nfs4")
+    assert unified._odirect_ok("/mnt/nfs/f") is False
+    monkeypatch.setattr(unified, "get_fs_type", lambda p: "ext4")
+    assert unified._odirect_ok("/data/f") is True
+    monkeypatch.setattr(unified, "get_fs_type", lambda p: "")  # unknown: allow
+    assert unified._odirect_ok("/x") is True
+
+
+def test_odirect_env_override(monkeypatch):
+    from fastsafetensors.copier import unified
+
+    monkeypatch.setattr(unified, "get_fs_type", lambda p: "nfs4")
+    monkeypatch.setenv("FASTSAFETENSORS_ODIRECT", "1")
+    assert unified._odirect_ok("/mnt/nfs/f") is True  # forced on
+    monkeypatch.setattr(unified, "get_fs_type", lambda p: "ext4")
+    monkeypatch.setenv("FASTSAFETENSORS_ODIRECT", "0")
+    assert unified._odirect_ok("/data/f") is False  # forced off
+
+
+# ---- chunk plans must fail loudly on copiers without set_chunk ----
+
+
+def test_chunk_plan_requires_set_chunk(input_files, framework):
+    if framework.get_name() != "pytorch":
+        pytest.skip("pytorch-only")
+    from fastsafetensors import SafeTensorsFileLoader, SafeTensorsMetadata
+
+    loader = SafeTensorsFileLoader(None, "cpu", nogds=True, framework="pytorch")
+    loader.add_filenames({0: [input_files[0]]})
+    meta = SafeTensorsMetadata.from_file(input_files[0], framework)
+    (chunk,) = meta.plan_chunks(meta.size_bytes)  # single whole-span chunk
+    loader.set_chunk_plan({input_files[0]: chunk})
+
+    class _NoChunkCopier:  # e.g. gds/dstorage: no set_chunk support
+        def __init__(self, *a, **k):
+            pass
+
+    loader.copier_constructor = lambda m, d, f: _NoChunkCopier()
+    with pytest.raises(NotImplementedError, match="set_chunk"):
+        loader.copy_files_to_device()
+
+
 # ---- runtime GDS -> nogds fallback ----