# Copyright (c) 2025, NVIDIA CORPORATION. All rights reserved.
# See file LICENSE for terms.
import os
import stat
from typing import Any, Optional
from kvikio._lib.mmap import InternalMmapHandle
from kvikio.cufile import IOFuture
[docs]
class Mmap:
"""Handle of a memory-mapped file"""
[docs]
def __init__(
self,
file_path: os.PathLike,
flags: str = "r",
initial_map_size: Optional[int] = None,
initial_map_offset: int = 0,
mode: int = stat.S_IRUSR | stat.S_IWUSR | stat.S_IRGRP | stat.S_IROTH,
map_flags: Optional[int] = None,
):
"""Construct a new memory-mapped file handle
Parameters
----------
file_path : os.PathLike
File path.
flags : str, optional
- ``r``: Open for reading (default)
- ``w``: (Not implemented yet) Open for writing, truncating the file first
- ``a``: (Not implemented yet) Open for writing, appending to the end of
file if it exists
- ``+``: (Not implemented yet) Open for updating (reading and writing)
initial_map_size : int, optional
Size in bytes of the mapped region. If not specified, map the region
starting from ``initial_map_offset`` to the end of file.
initial_map_offset : int, optional
File offset of the mapped region. Default is 0.
mode : int, optional
Access mode (permissions) to use if creating a new file. Default is
0644 (octal), 420 (decimal).
map_flags : int, optional
Flags to be passed to the system call ``mmap``. See `mmap(2)` for details.
"""
self._handle = InternalMmapHandle(
file_path, flags, initial_map_size, initial_map_offset, mode, map_flags
)
[docs]
def initial_map_size(self) -> int:
"""Size in bytes of the mapped region when the mapping handle was constructed
Returns
-------
int
Initial size of the mapped region.
"""
return self._handle.initial_map_size()
[docs]
def initial_map_offset(self) -> int:
"""File offset of the mapped region when the mapping handle was constructed
Returns
-------
int
Initial file offset of the mapped region.
"""
return self._handle.initial_map_offset()
[docs]
def file_size(self) -> int:
"""Get the file size if the file is open
Returns 0 if the file is closed.
Returns
-------
int
The file size in bytes.
"""
return self._handle.file_size()
[docs]
def close(self) -> None:
"""Close the mapping handle if it is open; do nothing otherwise
Unmaps the memory region and closes the underlying file descriptor.
"""
self._handle.close()
[docs]
def closed(self) -> bool:
"""Whether the mapping handle is closed
Returns
-------
bool
Boolean answer.
"""
return self._handle.closed()
[docs]
def read(self, buf: Any, size: Optional[int] = None, offset: int = 0) -> int:
"""Sequential read ``size`` bytes from the file to the destination buffer
``buf``
Parameters
----------
buf : buffer-like or array-like
Address of the host or device memory (destination buffer).
size : int, optional
Size in bytes to read. If not specified, read starts from ``offset``
to the end of file.
offset : int, optional
File offset. Default is 0.
Returns
-------
int
Number of bytes that have been read.
Raises
------
IndexError
If the read region specified by ``offset`` and ``size`` is outside the
initial region specified when the mapping handle was constructed.
RuntimeError
If the mapping handle is closed.
"""
return self._handle.read(buf, size, offset)
[docs]
def pread(
self,
buf: Any,
size: Optional[int] = None,
offset: int = 0,
task_size: Optional[int] = None,
) -> IOFuture:
"""Parallel read ``size`` bytes from the file to the destination buffer ``buf``
Parameters
----------
buf : buffer-like or array-like
Address of the host or device memory (destination buffer).
size : int, optional
Size in bytes to read. If not specified, read starts from ``offset``
to the end of file.
offset : int, optional
File offset. Default is 0.
task_size : int, optional
Size of each task in bytes for parallel execution. If None, uses
the default task size from :func:`kvikio.defaults.task_size`.
Returns
-------
IOFuture
Future that on completion returns the size of bytes that were successfully
read.
Raises
------
IndexError
If the read region specified by ``offset`` and ``size`` is outside the
initial region specified when the mapping handle was constructed.
RuntimeError
If the mapping handle is closed.
Notes
-----
The returned IOFuture object's ``get()`` should not be called after the lifetime
of the MmapHandle object ends. Otherwise, the behavior is undefined.
"""
return IOFuture(self._handle.pread(buf, size, offset, task_size))