vllm.v1.kv_offload.cpu.shared_offload_region

Classes:

• SharedOffloadRegion –

  Single mmap-backed memory region shared across all workers for a

SharedOffloadRegion

Single mmap-backed memory region shared across all workers for a vLLM instance. Workers coordinate via the filesystem: the first worker to open the file with O_EXCL becomes the creator and calls ftruncate; the rest open the existing file and wait until it reaches the expected size. Each worker then mmap()s the full file.

File path: /dev/shm/vllm_offload_{engine_id}.mmap. When a barrier is given, the path is unlinked once every worker has mapped the file, so the kernel reclaims the memory when the last worker exits, no matter how it exits; mappings taken before the unlink stay valid.

populate_shard=(index, count) pre-faults only this worker's contiguous share, for layouts where every worker maps the same bytes. The barrier also waits for all shards to finish pre-faulting before returning. num_segments splits backing files beneath the contiguous mapping.

Methods:

• create_kv_memoryview –

  Return a zero-copy memoryview over the entire KV buffer.

• create_next_canonical_view –

  Allocate a strided int8 view shared by all workers for one

• create_next_worker_view –

  Allocate a strided int8 view for this worker, one canonical tensor.

• iter_worker_slot_ranges –

  Yield this worker's slots as page-aligned (offset, length) ranges.

Source code in vllm/v1/kv_offload/cpu/shared_offload_region.py

class SharedOffloadRegion:
    """Single mmap-backed memory region shared across all workers for a
    vLLM instance.  Workers coordinate via the filesystem: the first worker
    to open the file with O_EXCL becomes the creator and calls ftruncate;
    the rest open the existing file and wait until it reaches the expected
    size.  Each worker then mmap()s the full file.

    File path: /dev/shm/vllm_offload_{engine_id}.mmap.  When a barrier is
    given, the path is unlinked once every worker has mapped the file, so
    the kernel reclaims the memory when the last worker exits, no matter
    how it exits; mappings taken before the unlink stay valid.

    ``populate_shard=(index, count)`` pre-faults only this worker's contiguous
    share, for layouts where every worker maps the same bytes. The barrier
    also waits for all shards to finish pre-faulting before returning.
    ``num_segments`` splits backing files beneath the contiguous mapping.
    """

    BLOCK_SIZE_ALIGNMENT: int = mmap.PAGESIZE

    def __init__(
        self,
        engine_id: str,
        num_chunks: int,
        rank: int | None,
        kv_bytes_per_chunk: int,
        cpu_page_size: int,
        barrier: Callable[[], None] | None = None,
        *,
        creator_memory_check: Callable[[int], None] | None = None,
        populate_shard: tuple[int, int] | None = None,
        num_segments: int = 1,
    ) -> None:
        self.page_size = mmap.PAGESIZE
        assert kv_bytes_per_chunk % self.page_size == 0

        self.num_chunks = num_chunks
        self._row_stride = kv_bytes_per_chunk
        self.total_size_bytes = self.num_chunks * self._row_stride
        self.cpu_page_size = cpu_page_size

        self.mmap_path = f"/dev/shm/vllm_offload_{engine_id}.mmap"
        num_segments = min(num_segments, self.total_size_bytes // _MIN_SEGMENT_BYTES)
        alignment = 2 << 20 if num_chunks == 1 else self._row_stride
        segment_bytes = round_up(
            cdiv(self.total_size_bytes, max(1, num_segments)), alignment
        )
        # (path, offset, length) per file.
        self._segments = [
            (
                f"{self.mmap_path}.{i}" if i else self.mmap_path,
                offset,
                min(segment_bytes, self.total_size_bytes - offset),
            )
            for i, offset in enumerate(range(0, self.total_size_bytes, segment_bytes))
        ]
        self._creator = False  # set True only if this worker creates the file
        self.rank = rank
        if rank is not None:
            # byte offset to this worker's first slot within each chunk row
            self._worker_offset = rank * cpu_page_size
            # exclusive upper bound for this worker's area within each row
            self._worker_area_end = (rank + 1) * cpu_page_size
        barriers_remaining = 2 if populate_shard is not None else 1
        try:
            try:
                self.fd: int | None = os.open(
                    self.mmap_path, os.O_CREAT | os.O_EXCL | os.O_RDWR, 0o600
                )
            except FileExistsError:
                # Joiner path — another worker won O_EXCL. Reopen and wait
                # for the file to reach expected size.
                self.fd = os.open(self.mmap_path, os.O_RDWR)
                _wait_for_file_size(self.fd, self._segments[0][2])
                logger.info("Opened existing mmap file %s", self.mmap_path)
            else:
                # Creator path. We won O_EXCL, so we own the file: any
                # failure here must clean up so concurrent joiners don't
                # land on a 0-byte stub and spin in _wait_for_file_size
                # for the full 30 s timeout.
                self._creator = True
                if creator_memory_check is not None:
                    creator_memory_check(self.total_size_bytes)
                check_shm_free_space(
                    self.total_size_bytes,
                    allocation_name="CPU KV offload shared region in /dev/shm",
                )
                for path, _, length in self._segments[1:]:
                    fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_RDWR, 0o600)
                    os.ftruncate(fd, length)
                    os.close(fd)
                # Joiners wait for the first file, so size it last.
                os.ftruncate(self.fd, self._segments[0][2])
                logger.info(
                    "Created mmap file %s (%.2f GB)",
                    self.mmap_path,
                    self.total_size_bytes / 1e9,
                )

            single = len(self._segments) == 1
            self.mmap_obj: mmap.mmap | None = None
            if single:
                self.mmap_obj = mmap.mmap(
                    self.fd,
                    self.total_size_bytes,
                    flags=mmap.MAP_SHARED,
                    prot=mmap.PROT_READ | mmap.PROT_WRITE,
                )
            else:
                reservation = os.memfd_create("vllm_offload_reservation")
                try:
                    os.ftruncate(reservation, self.total_size_bytes)
                    self.mmap_obj = mmap.mmap(
                        reservation,
                        self.total_size_bytes,
                        flags=mmap.MAP_SHARED,
                        prot=mmap.PROT_READ | mmap.PROT_WRITE,
                    )
                finally:
                    os.close(reservation)
                self._map_segments()
            if barrier is not None:
                barriers_remaining -= 1
                barrier()
                if self._creator:
                    self._unlink()
                    self._creator = False
                    logger.info("Unlinked mmap file %s", self.mmap_path)
            if populate_shard is not None:
                populate_write_fn = _get_populate_write_fn(self.mmap_obj)
                index, count = populate_shard
                share = round_up(cdiv(self.total_size_bytes, count), self.page_size)
                start = min(index * share, self.total_size_bytes)
                if start < self.total_size_bytes:
                    length = min(share, self.total_size_bytes - start)
                    populate_write_fn(self.mmap_obj, start, length)
                if barrier is not None:
                    barriers_remaining -= 1
                    barrier()
        except Exception:
            if self._creator:
                self._unlink()
                self._creator = False
            if hasattr(self, "mmap_obj") and self.mmap_obj is not None:
                # The prefault traceback may retain an exported buffer.
                with contextlib.suppress(BufferError):
                    self.mmap_obj.close()
                self.mmap_obj = None
            if hasattr(self, "fd") and self.fd is not None:
                os.close(self.fd)
                self.fd = None
            # Join remaining barriers so peers do not wait until the collective
            # times out. A barrier failure must not replace the setup error.
            if barrier is not None:
                for _ in range(barriers_remaining):
                    try:
                        barrier()
                    except Exception:
                        logger.warning(
                            "Failed to release peers waiting at the mmap barrier",
                            exc_info=True,
                        )
            raise

        self._base = torch.frombuffer(memoryview(self.mmap_obj), dtype=torch.int8)
        self._views: list[torch.Tensor] = []
        self._canonical_offset = 0
        self.is_pinned: bool = False
        self.pinned_addresses: list[int] = []

        if populate_shard is not None:
            return
        populate_write_fn = _get_populate_write_fn(self.mmap_obj)

        if rank is not None:
            # Populate only this worker's pages (one slot per chunk row).
            _t0 = time.perf_counter()
            for offset, length in self.iter_worker_slot_ranges():
                populate_write_fn(self.mmap_obj, offset, length)
            logger.debug(
                "MADV_POPULATE_WRITE loop: %d chunks in %.3f s",
                num_chunks,
                time.perf_counter() - _t0,
            )
        else:
            # No rank — populate the entire shared region in one call.
            _t0 = time.perf_counter()
            populate_write_fn(self.mmap_obj, 0, self.total_size_bytes)
            logger.debug(
                "MADV_POPULATE_WRITE entire region: %.3f s", time.perf_counter() - _t0
            )

    def _map_segments(self) -> None:
        """Replace the anonymous mapping with the segment files, back to back."""
        assert self.mmap_obj is not None and self.fd is not None
        base = ctypes.addressof(ctypes.c_char.from_buffer(self.mmap_obj))
        for path, offset, length in self._segments:
            fd = self.fd if offset == 0 else os.open(path, os.O_RDWR)
            address = _libc_mmap()(
                base + offset,
                length,
                mmap.PROT_READ | mmap.PROT_WRITE,
                mmap.MAP_SHARED | _MAP_FIXED,
                fd,
                0,
            )
            if fd != self.fd:
                os.close(fd)
            if address != base + offset:
                raise OSError(ctypes.get_errno(), f"Failed to map {path}")

    def _unlink(self) -> None:
        for path, _, _ in self._segments:
            with contextlib.suppress(FileNotFoundError):
                os.unlink(path)

    def iter_worker_slot_ranges(self) -> Iterator[tuple[int, int]]:
        """Yield this worker's slots as page-aligned (offset, length) ranges."""
        assert self.rank is not None
        worker_offset = self.rank * self.cpu_page_size
        first = 0
        if len(self._segments) > 1:
            # Start each worker on a different file.
            first = (
                self.rank * self.num_chunks // (self._row_stride // self.cpu_page_size)
            )
        for i in range(self.num_chunks):
            row = (first + i) % self.num_chunks
            start = row * self._row_stride + worker_offset
            aligned = start - start % self.page_size
            end = round_up(start + self.cpu_page_size, self.page_size)
            yield aligned, end - aligned

    @property
    def base_tensor(self) -> torch.Tensor:
        if self._base is None:
            raise RuntimeError("Shared offload region has been released.")
        return self._base

    def create_next_worker_view(self, tensor_page_size: int) -> torch.Tensor:
        """Allocate a strided int8 view for this worker, one canonical tensor.

        Must be called once per canonical tensor. The full mmap layout is:

            worker0_chunk0 | worker1_chunk0 | ... | worker{M-1}_chunk0
            worker0_chunk1 | worker1_chunk1 | ... | worker{M-1}_chunk1
            ...

        Each worker_chunk cell is cpu_page_size bytes and holds all canonical
        tensors for that worker and chunk concatenated:
            [ tensor0_data | tensor1_data | ... | tensor{L-1}_data ]

        Consecutive rows are separated by row_stride = cpu_page_size * M.

        Returns an int8 tensor of shape (num_chunks, tensor_page_size) with stride
        (row_stride, 1).  Using int8 keeps stride == bytes, so swap_blocks
        address arithmetic works without any dtype conversion.

        Args:
            tensor_page_size: Bytes per chunk for this tensor.

        """
        assert self.rank is not None
        new_offset = self._worker_offset + tensor_page_size
        assert new_offset <= self._worker_area_end, (
            f"Worker offset {new_offset} exceeds worker area end "
            f"{self._worker_area_end} (overflowed by "
            f"{new_offset - self._worker_area_end} bytes)"
        )
        worker_layer_view = torch.as_strided(
            self._base,
            size=(self.num_chunks, tensor_page_size),
            stride=(self._row_stride, 1),
            storage_offset=self._worker_offset,
        )
        self._worker_offset = new_offset
        self._views.append(worker_layer_view)
        return worker_layer_view

    def create_next_canonical_view(self, tensor_page_size: int) -> torch.Tensor:
        """Allocate a strided int8 view shared by all workers for one
        canonical tensor (canonical layout).

        Must be called once per canonical tensor, instead of
        create_next_worker_view. The full mmap layout is:

            |<-------- canonical area ------->|<-------- unused ------->|
            |  all workers share this area    |                         |
            |                                 |                         |
            | [ canonical_t0 | canonical_t1 ] |                         |
            | [ canonical_t0 | canonical_t1 ] |                         |
            | [ canonical_t0 | canonical_t1 ] |                         |
            ^                ^
            _canonical_offset=0, then advances by each tensor's size

        Each canonical_t{i} cell is that tensor's canonical page for the
        chunk. Canonical areas are carved consecutively from the start of
        each chunk row; consecutive rows are separated by row_stride. Every
        worker gets the identical byte ranges and writes only its disjoint
        bytes within them, as described by its canonical mappings — unlike
        create_next_worker_view, which gives each worker a private
        cpu_page_size slot per row.

        The trailing unused bytes exist only when the canonical pages sum to
        less than row_stride: page-alignment padding of the row, or
        deduplication of KV replicated across workers (e.g. the MLA latent),
        where one canonical copy replaces world_size worker copies.

        Args:
            tensor_page_size: Canonical bytes per chunk for this tensor.

        """
        new_offset = self._canonical_offset + tensor_page_size
        assert new_offset <= self._row_stride
        view = torch.as_strided(
            self._base,
            size=(self.num_chunks, tensor_page_size),
            stride=(self._row_stride, 1),
            storage_offset=self._canonical_offset,
        )
        self._canonical_offset = new_offset
        self._views.append(view)
        return view

    def create_kv_memoryview(self) -> memoryview:
        """Return a zero-copy memoryview over the entire KV buffer.

        Shape: (num_chunks, row_stride_bytes). Secondary tiers address
        chunk *b* as ``view[b]``.
        """
        kv_tensor = self._base.view(self.num_chunks, self._row_stride)
        np_arr = kv_tensor.numpy()
        assert np_arr.ctypes.data == self._base.data_ptr(), (
            "view()/numpy() created a copy instead of sharing the mmap buffer; "
            "secondary tiers require zero-copy access to primary KV data"
        )
        return memoryview(np_arr)

    def cleanup(self) -> None:
        if self.is_pinned and self._base is not None:
            base_ptr = self._base.data_ptr()
            addresses = self.pinned_addresses or [base_ptr]
            for address in reversed(addresses):
                host_unregister(address)
            self.pinned_addresses.clear()
            self.is_pinned = False
        # Release views before _base: each view holds a _base reference and a
        # direct StorageImpl reference.  Freeing views first lets both refcounts
        # drop so the storage (which holds the mmap_obj buffer export) is freed
        # before mmap_obj.close() is called below.
        if self._views is not None:
            self._views.clear()
        self._base = None
        if self.mmap_obj:
            retained = False
            try:
                self.mmap_obj.close()
            except BufferError:
                # A native user still exports this mapping, so unmapping would
                # let the NIC write into reused address space. Keep it mapped;
                # dropping the fd and the pathname below is still safe.
                retained = True
                logger.warning(
                    "Keeping %s mapped: buffers are still exported to native users",
                    self.mmap_path,
                )
            except Exception:
                logger.warning("Failed to close mmap_obj", exc_info=True)
            if not retained:
                self.mmap_obj = None
        if self.fd is not None:
            try:
                os.close(self.fd)
            except Exception:
                logger.warning("Failed to close fd %s", self.fd, exc_info=True)
            self.fd = None
        if self._creator and getattr(self, "mmap_path", None):
            try:
                self._unlink()
                logger.info("Removed mmap file %s", self.mmap_path)
            except Exception:
                logger.warning(
                    "Failed to unlink path %s", self.mmap_path, exc_info=True
                )
            self._creator = False

_map_segments()

Replace the anonymous mapping with the segment files, back to back.

Source code in vllm/v1/kv_offload/cpu/shared_offload_region.py

def _map_segments(self) -> None:
    """Replace the anonymous mapping with the segment files, back to back."""
    assert self.mmap_obj is not None and self.fd is not None
    base = ctypes.addressof(ctypes.c_char.from_buffer(self.mmap_obj))
    for path, offset, length in self._segments:
        fd = self.fd if offset == 0 else os.open(path, os.O_RDWR)
        address = _libc_mmap()(
            base + offset,
            length,
            mmap.PROT_READ | mmap.PROT_WRITE,
            mmap.MAP_SHARED | _MAP_FIXED,
            fd,
            0,
        )
        if fd != self.fd:
            os.close(fd)
        if address != base + offset:
            raise OSError(ctypes.get_errno(), f"Failed to map {path}")

create_kv_memoryview()

Return a zero-copy memoryview over the entire KV buffer.

Shape: (num_chunks, row_stride_bytes). Secondary tiers address chunk b as view[b].

Source code in vllm/v1/kv_offload/cpu/shared_offload_region.py

def create_kv_memoryview(self) -> memoryview:
    """Return a zero-copy memoryview over the entire KV buffer.

    Shape: (num_chunks, row_stride_bytes). Secondary tiers address
    chunk *b* as ``view[b]``.
    """
    kv_tensor = self._base.view(self.num_chunks, self._row_stride)
    np_arr = kv_tensor.numpy()
    assert np_arr.ctypes.data == self._base.data_ptr(), (
        "view()/numpy() created a copy instead of sharing the mmap buffer; "
        "secondary tiers require zero-copy access to primary KV data"
    )
    return memoryview(np_arr)

create_next_canonical_view(tensor_page_size)

Allocate a strided int8 view shared by all workers for one canonical tensor (canonical layout).

Must be called once per canonical tensor, instead of create_next_worker_view. The full mmap layout is:

|<-------- canonical area ------->|<-------- unused ------->|
|  all workers share this area    |                         |
|                                 |                         |
| [ canonical_t0 | canonical_t1 ] |                         |
| [ canonical_t0 | canonical_t1 ] |                         |
| [ canonical_t0 | canonical_t1 ] |                         |
^                ^
_canonical_offset=0, then advances by each tensor's size

Each canonical_t{i} cell is that tensor's canonical page for the chunk. Canonical areas are carved consecutively from the start of each chunk row; consecutive rows are separated by row_stride. Every worker gets the identical byte ranges and writes only its disjoint bytes within them, as described by its canonical mappings — unlike create_next_worker_view, which gives each worker a private cpu_page_size slot per row.

The trailing unused bytes exist only when the canonical pages sum to less than row_stride: page-alignment padding of the row, or deduplication of KV replicated across workers (e.g. the MLA latent), where one canonical copy replaces world_size worker copies.

Parameters:

• tensor_page_size

  (int) –

  Canonical bytes per chunk for this tensor.

Source code in vllm/v1/kv_offload/cpu/shared_offload_region.py

def create_next_canonical_view(self, tensor_page_size: int) -> torch.Tensor:
    """Allocate a strided int8 view shared by all workers for one
    canonical tensor (canonical layout).

    Must be called once per canonical tensor, instead of
    create_next_worker_view. The full mmap layout is:

        |<-------- canonical area ------->|<-------- unused ------->|
        |  all workers share this area    |                         |
        |                                 |                         |
        | [ canonical_t0 | canonical_t1 ] |                         |
        | [ canonical_t0 | canonical_t1 ] |                         |
        | [ canonical_t0 | canonical_t1 ] |                         |
        ^                ^
        _canonical_offset=0, then advances by each tensor's size

    Each canonical_t{i} cell is that tensor's canonical page for the
    chunk. Canonical areas are carved consecutively from the start of
    each chunk row; consecutive rows are separated by row_stride. Every
    worker gets the identical byte ranges and writes only its disjoint
    bytes within them, as described by its canonical mappings — unlike
    create_next_worker_view, which gives each worker a private
    cpu_page_size slot per row.

    The trailing unused bytes exist only when the canonical pages sum to
    less than row_stride: page-alignment padding of the row, or
    deduplication of KV replicated across workers (e.g. the MLA latent),
    where one canonical copy replaces world_size worker copies.

    Args:
        tensor_page_size: Canonical bytes per chunk for this tensor.

    """
    new_offset = self._canonical_offset + tensor_page_size
    assert new_offset <= self._row_stride
    view = torch.as_strided(
        self._base,
        size=(self.num_chunks, tensor_page_size),
        stride=(self._row_stride, 1),
        storage_offset=self._canonical_offset,
    )
    self._canonical_offset = new_offset
    self._views.append(view)
    return view

create_next_worker_view(tensor_page_size)

Allocate a strided int8 view for this worker, one canonical tensor.

Must be called once per canonical tensor. The full mmap layout is:

worker0_chunk0 | worker1_chunk0 | ... | worker{M-1}_chunk0
worker0_chunk1 | worker1_chunk1 | ... | worker{M-1}_chunk1
...

Each worker_chunk cell is cpu_page_size bytes and holds all canonical tensors for that worker and chunk concatenated: [ tensor0_data | tensor1_data | ... | tensor{L-1}_data ]

Consecutive rows are separated by row_stride = cpu_page_size * M.

Returns an int8 tensor of shape (num_chunks, tensor_page_size) with stride (row_stride, 1). Using int8 keeps stride == bytes, so swap_blocks address arithmetic works without any dtype conversion.

Parameters:

• tensor_page_size

  (int) –

  Bytes per chunk for this tensor.

Source code in vllm/v1/kv_offload/cpu/shared_offload_region.py

def create_next_worker_view(self, tensor_page_size: int) -> torch.Tensor:
    """Allocate a strided int8 view for this worker, one canonical tensor.

    Must be called once per canonical tensor. The full mmap layout is:

        worker0_chunk0 | worker1_chunk0 | ... | worker{M-1}_chunk0
        worker0_chunk1 | worker1_chunk1 | ... | worker{M-1}_chunk1
        ...

    Each worker_chunk cell is cpu_page_size bytes and holds all canonical
    tensors for that worker and chunk concatenated:
        [ tensor0_data | tensor1_data | ... | tensor{L-1}_data ]

    Consecutive rows are separated by row_stride = cpu_page_size * M.

    Returns an int8 tensor of shape (num_chunks, tensor_page_size) with stride
    (row_stride, 1).  Using int8 keeps stride == bytes, so swap_blocks
    address arithmetic works without any dtype conversion.

    Args:
        tensor_page_size: Bytes per chunk for this tensor.

    """
    assert self.rank is not None
    new_offset = self._worker_offset + tensor_page_size
    assert new_offset <= self._worker_area_end, (
        f"Worker offset {new_offset} exceeds worker area end "
        f"{self._worker_area_end} (overflowed by "
        f"{new_offset - self._worker_area_end} bytes)"
    )
    worker_layer_view = torch.as_strided(
        self._base,
        size=(self.num_chunks, tensor_page_size),
        stride=(self._row_stride, 1),
        storage_offset=self._worker_offset,
    )
    self._worker_offset = new_offset
    self._views.append(worker_layer_view)
    return worker_layer_view

iter_worker_slot_ranges()

Yield this worker's slots as page-aligned (offset, length) ranges.

Source code in vllm/v1/kv_offload/cpu/shared_offload_region.py

def iter_worker_slot_ranges(self) -> Iterator[tuple[int, int]]:
    """Yield this worker's slots as page-aligned (offset, length) ranges."""
    assert self.rank is not None
    worker_offset = self.rank * self.cpu_page_size
    first = 0
    if len(self._segments) > 1:
        # Start each worker on a different file.
        first = (
            self.rank * self.num_chunks // (self._row_stride // self.cpu_page_size)
        )
    for i in range(self.num_chunks):
        row = (first + i) % self.num_chunks
        start = row * self._row_stride + worker_offset
        aligned = start - start % self.page_size
        end = round_up(start + self.cpu_page_size, self.page_size)
        yield aligned, end - aligned

_get_populate_write_fn(mmap_obj)

Select the pre-faulting method once for this mmap.

Source code in vllm/v1/kv_offload/cpu/shared_offload_region.py

def _get_populate_write_fn(
    mmap_obj: mmap.mmap,
) -> Callable[[mmap.mmap, int, int], None]:
    """Select the pre-faulting method once for this mmap."""
    try:
        _madvise_populate_write(mmap_obj, 0, mmap.PAGESIZE)
    except OSError as e:
        if e.errno != errno.EINVAL:
            raise
        logger.warning(
            "MADV_POPULATE_WRITE is not supported; falling back to per-page "
            "writes for mmap pre-population. Startup may be slower."
        )
        return _fallback_populate_write
    return _madvise_populate_write

_wait_for_file_size(fd, expected_size, timeout=30.0)

Spin-wait until the file reaches expected_size (creator truncated it).

Source code in vllm/v1/kv_offload/cpu/shared_offload_region.py

def _wait_for_file_size(fd: int, expected_size: int, timeout: float = 30.0) -> None:
    """Spin-wait until the file reaches expected_size (creator truncated it)."""
    deadline = time.monotonic() + timeout
    while True:
        if os.fstat(fd).st_size >= expected_size:
            return
        if os.fstat(fd).st_nlink == 0:
            raise RuntimeError(
                "Shared offload region creator failed during initialization."
            )
        if time.monotonic() > deadline:
            raise TimeoutError(
                f"Timed out waiting for mmap file to reach {expected_size} bytes"
            )
        time.sleep(0.005)