From cfaa52e286ee6038713dbf9974b4ed79f54dec45 Mon Sep 17 00:00:00 2001 From: Jiejing Zhang Date: Tue, 16 Jun 2026 23:19:43 -0600 Subject: [PATCH 1/2] feat(hipfile/python): add async stream I/O bindings Expose the hipFile async API to the Python bindings. The C API (hipFileReadAsync / hipFileWriteAsync / hipFileStreamRegister / hipFileStreamDeregister) already exists; only the Python layer was missing -- the bindings previously had just the hipFileAsyncNotSupported enum. - _chipfile.pxd: declare hipStream_t + the four async functions. - _hipfile.pyx: AsyncIOHandle cdef class that owns the in/out C slots (size / file_offset / buffer_offset / bytes_done) so their addresses stay valid until the stream completes. The driver dereferences these pointers AFTER the async call returns (the transfer finishes on stream sync), so a plain Cython stack-local passed by address would be written back to a dead address -- bytes_done reads 0 and later submits can hit hipFileInvalidValue. Plus read/write async wrappers, stream register/deregister, and a supports_async() probe. - file.py: FileHandle.read_async / write_async returning an AsyncIOHandle (keep alive past stream sync, then read bytes_done), a Stream context manager, and supports_async(). - __init__.py: export Stream and supports_async. Mirrors the cuFile async pattern. Validated on gfx942 / ROCm 7.2: build + import, supports_async() == True, and a GPU async write->read round-trip (bytes_done == 4096, data byte-identical). --- projects/hipfile/python/hipfile/__init__.py | 4 +- projects/hipfile/python/hipfile/_chipfile.pxd | 22 +++ projects/hipfile/python/hipfile/_hipfile.pyx | 143 ++++++++++++++++ projects/hipfile/python/hipfile/file.py | 153 ++++++++++++++++++ 4 files changed, 321 insertions(+), 1 deletion(-) diff --git a/projects/hipfile/python/hipfile/__init__.py b/projects/hipfile/python/hipfile/__init__.py index bfe4d1235ea..eda133d6cc8 100644 --- a/projects/hipfile/python/hipfile/__init__.py +++ b/projects/hipfile/python/hipfile/__init__.py @@ -10,13 +10,15 @@ from hipfile.driver import Driver from hipfile.enums import FileHandleType, OpError from hipfile.error import HipFileException -from hipfile.file import FileHandle +from hipfile.file import FileHandle, Stream, supports_async from hipfile.properties import driver_get_properties, get_version __all__ = [ "__version__", "Driver", "FileHandle", + "Stream", + "supports_async", "Buffer", "HipFileException", "FileHandleType", diff --git a/projects/hipfile/python/hipfile/_chipfile.pxd b/projects/hipfile/python/hipfile/_chipfile.pxd index 748d4c10144..57a2b6d435e 100644 --- a/projects/hipfile/python/hipfile/_chipfile.pxd +++ b/projects/hipfile/python/hipfile/_chipfile.pxd @@ -19,6 +19,7 @@ cdef extern from "hip/hip_runtime_api.h": hipSuccess = 0 hipError_t hipPeekAtLastError() nogil + ctypedef void *hipStream_t # opaque CUDA/HIP stream handle # --------------------------------------------------------------------------- @@ -177,6 +178,27 @@ cdef extern from "hipfile.h": size_t size, hoff_t file_offset, hoff_t buffer_offset) nogil + # Asynchronous (stream-attached) I/O. The size/file_off/buf_off + # pointers are in/out and ``bytes_done_p`` receives the transferred + # byte count once the async op completes on ``stream``; the caller + # must keep all four storage slots alive and sync on the stream + # before reading them (see AsyncIOHandle in _hipfile.pyx). + hipFileError_t hipFileReadAsync(hipFileHandle_t fh, void *buffer_base, + size_t *size_p, hoff_t *file_offset_p, + hoff_t *buffer_offset_p, + ssize_t *bytes_read_p, + hipStream_t stream) nogil + hipFileError_t hipFileWriteAsync(hipFileHandle_t fh, void *buffer_base, + size_t *size_p, hoff_t *file_offset_p, + hoff_t *buffer_offset_p, + ssize_t *bytes_written_p, + hipStream_t stream) nogil + + # Stream registration — required before any *Async call on a stream. + hipFileError_t hipFileStreamRegister(hipStream_t stream, + unsigned flags) nogil + hipFileError_t hipFileStreamDeregister(hipStream_t stream) nogil + # Driver lifecycle hipFileError_t hipFileDriverOpen() nogil hipFileError_t hipFileDriverClose() nogil diff --git a/projects/hipfile/python/hipfile/_hipfile.pyx b/projects/hipfile/python/hipfile/_hipfile.pyx index 11c747c0c5b..8d8a1bb7b5c 100644 --- a/projects/hipfile/python/hipfile/_hipfile.pyx +++ b/projects/hipfile/python/hipfile/_hipfile.pyx @@ -283,6 +283,149 @@ def hipFileWrite(uintptr_t handle, uintptr_t buffer_base, size_t size, return (ret, extra) +# --------------------------------------------------------------------------- +# Asynchronous (stream-attached) I/O +# --------------------------------------------------------------------------- +# +# ``hipFileReadAsync`` / ``hipFileWriteAsync`` take pointer arguments +# (size, file/buffer offset, bytes-done) that the driver dereferences +# *after* the C call returns — the transfer actually completes when the +# stream is synchronised. Passing the address of a Cython stack-local +# would therefore write back to a dead address (``bytes_done`` reads as +# 0; later submits can hit ``hipFileInvalidValue``). ``AsyncIOHandle`` +# owns those slots as C members so their addresses stay valid for the +# object's lifetime; the caller keeps the handle alive past the stream +# sync and then reads ``bytes_done``. Mirrors the cuFile async pattern. + + +cdef class AsyncIOHandle: + """In/out C storage for one async submit. + + Keep this object alive until the stream the I/O was submitted to has + been synchronised (``Stream.synchronize`` / a recorded ``Event``), + then read :attr:`bytes_done`. + """ + + cdef size_t _size + cdef _c.hoff_t _file_off + cdef _c.hoff_t _buf_off + cdef ssize_t _bytes_done + + def __cinit__(self, size_t size, _c.hoff_t file_offset, + _c.hoff_t buffer_offset): + self._size = size + self._file_off = file_offset + self._buf_off = buffer_offset + self._bytes_done = 0 + + @property + def bytes_done(self): + """Bytes transferred, valid only after the stream has synced.""" + return self._bytes_done + + @property + def size(self): + return self._size + + @property + def file_offset(self): + return self._file_off + + @property + def buffer_offset(self): + return self._buf_off + + +def hipFileReadAsync(uintptr_t handle, uintptr_t buffer_base, + AsyncIOHandle io not None, uintptr_t stream_handle): + """Wrapper for ``hipFileReadAsync``. + + Submits the read to ``stream_handle`` and returns immediately. + ``io`` carries the in/out slots; keep it alive and sync on the + stream before reading ``io.bytes_done``. + + Returns ``(err, extra)``: ``err == 0`` on a successful submit; + otherwise ``err`` is ``hipFileOpError_t`` and ``extra`` is + ``hipError_t`` (when ``err == hipFileHipDriverError``) or ``errno``. + """ + cdef _c.hipFileError_t e + cdef int extra = 0 + cdef size_t *size_p = &io._size + cdef _c.hoff_t *foff_p = &io._file_off + cdef _c.hoff_t *boff_p = &io._buf_off + cdef ssize_t *done_p = &io._bytes_done + with nogil: + e = _c.hipFileReadAsync(<_c.hipFileHandle_t>handle, + buffer_base, + size_p, foff_p, boff_p, done_p, + <_c.hipStream_t>stream_handle) + if e.err != _c.hipFileSuccess: + if e.err == _c.hipFileHipDriverError: + extra = e.hip_drv_err + else: + extra = errno + return (e.err, extra) + + +def hipFileWriteAsync(uintptr_t handle, uintptr_t buffer_base, + AsyncIOHandle io not None, uintptr_t stream_handle): + """Wrapper for ``hipFileWriteAsync``. See :func:`hipFileReadAsync` + for argument lifetime and return semantics.""" + cdef _c.hipFileError_t e + cdef int extra = 0 + cdef size_t *size_p = &io._size + cdef _c.hoff_t *foff_p = &io._file_off + cdef _c.hoff_t *boff_p = &io._buf_off + cdef ssize_t *done_p = &io._bytes_done + with nogil: + e = _c.hipFileWriteAsync(<_c.hipFileHandle_t>handle, + buffer_base, + size_p, foff_p, boff_p, done_p, + <_c.hipStream_t>stream_handle) + if e.err != _c.hipFileSuccess: + if e.err == _c.hipFileHipDriverError: + extra = e.hip_drv_err + else: + extra = errno + return (e.err, extra) + + +def hipFileStreamRegister(uintptr_t stream_handle, unsigned flags=0): + """Wrapper for ``hipFileStreamRegister``. Register a CUDA/HIP stream + before submitting any async I/O to it. Returns the ``(err, extra)`` + tuple (``err == 0`` on success).""" + cdef _c.hipFileError_t e + cdef int extra = 0 + with nogil: + e = _c.hipFileStreamRegister(<_c.hipStream_t>stream_handle, flags) + if e.err == _c.hipFileHipDriverError: + extra = e.hip_drv_err + return (e.err, extra) + + +def hipFileStreamDeregister(uintptr_t stream_handle): + """Wrapper for ``hipFileStreamDeregister``. Returns ``(err, extra)``.""" + cdef _c.hipFileError_t e + cdef int extra = 0 + with nogil: + e = _c.hipFileStreamDeregister(<_c.hipStream_t>stream_handle) + if e.err == _c.hipFileHipDriverError: + extra = e.hip_drv_err + return (e.err, extra) + + +def supports_async(): + """Return ``True`` if the loaded libhipfile implements the async + stream API. Probes ``hipFileStreamRegister`` on a null stream and + treats anything other than ``hipFileAsyncNotSupported`` as + supported (a null stream may be rejected with a different code on a + driver that *does* implement the API).""" + cdef _c.hipFileError_t e + with nogil: + e = _c.hipFileStreamRegister(<_c.hipStream_t>0, 0) + return e.err != _c.hipFileAsyncNotSupported + + # --------------------------------------------------------------------------- # Driver properties # --------------------------------------------------------------------------- diff --git a/projects/hipfile/python/hipfile/file.py b/projects/hipfile/python/hipfile/file.py index fb258b0d742..7d30075c2be 100644 --- a/projects/hipfile/python/hipfile/file.py +++ b/projects/hipfile/python/hipfile/file.py @@ -8,10 +8,16 @@ from typing import TYPE_CHECKING from hipfile._hipfile import ( # pylint: disable=E0401,E0611 + AsyncIOHandle, hipFileHandleRegister, hipFileHandleDeregister, hipFileRead, + hipFileReadAsync, + hipFileStreamDeregister, + hipFileStreamRegister, hipFileWrite, + hipFileWriteAsync, + supports_async as _supports_async, ) from hipfile.enums import FileHandleType from hipfile.error import HipFileException @@ -265,3 +271,150 @@ def write( # Otherwise, extra_err is 0. raise HipFileException(-bytes_written, extra_err) return bytes_written + + def read_async( + self, + buffer: Buffer, + size: int, + file_offset: int, + buffer_offset: int, + stream: int, + ) -> AsyncIOHandle: + """Submit an asynchronous read into a GPU buffer on *stream*. + + The read is queued on the CUDA/HIP stream and this returns + immediately. The returned :class:`AsyncIOHandle` owns the in/out + slots the driver fills in when the I/O completes; the caller MUST + keep it alive until the stream has been synchronised (e.g. a + recorded ``Event`` the consumer waits on), then read + ``handle.bytes_done``. + + The stream must have been registered first (see :class:`Stream`). + + Parameters + ---------- + buffer : Buffer + GPU buffer to read into. + size : int + Number of bytes to read. + file_offset : int + Byte offset within the file to start reading from. + buffer_offset : int + Byte offset within the GPU buffer to read into. + stream : int + Opaque CUDA/HIP stream handle (e.g. + ``torch.cuda.Stream.cuda_stream``). + + Returns + ------- + AsyncIOHandle + Keep alive past stream sync; then read ``bytes_done``. + + Raises + ------ + RuntimeError + If the file handle is not open. + OSError + On a system-level I/O error (wraps ``errno``). + HipFileException + On a hipFile or HIP driver error at submit time. + """ + if self._handle is None: + raise RuntimeError("The FileHandle is not open.") + io = AsyncIOHandle(size, file_offset, buffer_offset) + err, extra_err = hipFileReadAsync(self._handle, buffer.ptr, io, stream) + if err != 0: + # err is hipFileOpError_t; extra_err is hipError_t when err == + # hipFileHipDriverError, else errno (advisory). + raise HipFileException(err, extra_err) + return io + + def write_async( + self, + buffer: Buffer, + size: int, + file_offset: int, + buffer_offset: int, + stream: int, + ) -> AsyncIOHandle: + """Submit an asynchronous write from a GPU buffer on *stream*. + + See :meth:`read_async` for argument lifetime and return + semantics. + """ + if self._handle is None: + raise RuntimeError("The FileHandle is not open.") + io = AsyncIOHandle(size, file_offset, buffer_offset) + err, extra_err = hipFileWriteAsync(self._handle, buffer.ptr, io, stream) + if err != 0: + raise HipFileException(err, extra_err) + return io + + +class Stream: + """Register a CUDA/HIP stream with hipFile for asynchronous I/O. + + A stream must be registered before any ``read_async`` / ``write_async`` + targets it, and deregistered at shutdown. Supports the context-manager + protocol:: + + with Stream(torch_stream.cuda_stream) as st: + handle = fh.read_async(buf, size, 0, 0, st.handle) + event.record(torch_stream) + consumer_stream.wait_event(event) + # keep `handle` alive until the consumer has waited, then + # handle.bytes_done is valid. + """ + + def __init__(self, stream: int, flags: int = 0) -> None: + """Wrap an opaque CUDA/HIP stream handle (not yet registered).""" + self._stream = int(stream) + self._flags = int(flags) + self._registered = False + + @property + def handle(self) -> int: + """The opaque stream handle.""" + return self._stream + + @property + def registered(self) -> bool: + """Whether the stream is currently registered.""" + return self._registered + + def register(self) -> None: + """Register the stream. Idempotent.""" + if self._registered: + return + err, extra = hipFileStreamRegister(self._stream, self._flags) + if err != 0: + raise HipFileException(err, extra) + self._registered = True + + def deregister(self) -> None: + """Deregister the stream. Idempotent.""" + if not self._registered: + return + self._registered = False + err, extra = hipFileStreamDeregister(self._stream) + if err != 0: + raise HipFileException(err, extra) + + def __enter__(self) -> Stream: + self.register() + return self + + def __exit__( + self, + exc_type: type[BaseException] | None, + exc_value: BaseException | None, + traceback: TracebackType | None, + ) -> None: + self.deregister() + + +def supports_async() -> bool: + """Return ``True`` if the loaded hipFile library implements the + asynchronous stream I/O API, ``False`` if it is synchronous-only + (callers should then fall back to ``read`` / ``write``).""" + return bool(_supports_async()) From 90cb0c144b6e689f26614620479bce58d9c40500 Mon Sep 17 00:00:00 2001 From: Jiejing Zhang Date: Thu, 9 Jul 2026 17:47:36 +0000 Subject: [PATCH 2/2] hipfile(python): address async-bindings review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - errno only for POSIX/C errors (err == -1) in the async read/write wrappers, and also capture it for stream register/deregister; other hipFileOpError_t codes leave extra=0 (errno would be stale). Mirrors the sync path. - read_async/write_async now raise OSError on err == -1 (consistent with read()/write() and the hipFile error contract), HipFileException otherwise. - supports_async(): deregister the default stream if the probe registered it, so the probe leaves no permanent registration behind. - Stream.deregister(): only clear _registered after a successful deregister. - AsyncIOHandle: add size/file_offset/buffer_offset setters (async API allows modifying these after submission); fix docstring (no Stream.synchronize — synchronise the underlying HIP/CUDA stream / wait on an event). --- projects/hipfile/python/hipfile/_hipfile.pyx | 36 +++++++++++++++++--- projects/hipfile/python/hipfile/file.py | 10 ++++-- 2 files changed, 40 insertions(+), 6 deletions(-) diff --git a/projects/hipfile/python/hipfile/_hipfile.pyx b/projects/hipfile/python/hipfile/_hipfile.pyx index 8d8a1bb7b5c..f19510d7109 100644 --- a/projects/hipfile/python/hipfile/_hipfile.pyx +++ b/projects/hipfile/python/hipfile/_hipfile.pyx @@ -302,8 +302,13 @@ cdef class AsyncIOHandle: """In/out C storage for one async submit. Keep this object alive until the stream the I/O was submitted to has - been synchronised (``Stream.synchronize`` / a recorded ``Event``), - then read :attr:`bytes_done`. + been synchronised (synchronise the underlying HIP/CUDA stream, e.g. + ``hipStreamSynchronize`` / ``torch.cuda.Stream.synchronize``, or wait on + a recorded event), then read :attr:`bytes_done`. + + ``size`` / ``file_offset`` / ``buffer_offset`` are writable: the async API + allows setting them after submission (when not known at submit time). The + driver reads the underlying C slots when the op runs on the stream. """ cdef size_t _size @@ -327,14 +332,26 @@ cdef class AsyncIOHandle: def size(self): return self._size + @size.setter + def size(self, value): + self._size = value + @property def file_offset(self): return self._file_off + @file_offset.setter + def file_offset(self, value): + self._file_off = value + @property def buffer_offset(self): return self._buf_off + @buffer_offset.setter + def buffer_offset(self, value): + self._buf_off = value + def hipFileReadAsync(uintptr_t handle, uintptr_t buffer_base, AsyncIOHandle io not None, uintptr_t stream_handle): @@ -362,7 +379,8 @@ def hipFileReadAsync(uintptr_t handle, uintptr_t buffer_base, if e.err != _c.hipFileSuccess: if e.err == _c.hipFileHipDriverError: extra = e.hip_drv_err - else: + elif e.err == -1: + # errno is only meaningful for a POSIX/C error (err == -1). extra = errno return (e.err, extra) @@ -385,7 +403,8 @@ def hipFileWriteAsync(uintptr_t handle, uintptr_t buffer_base, if e.err != _c.hipFileSuccess: if e.err == _c.hipFileHipDriverError: extra = e.hip_drv_err - else: + elif e.err == -1: + # errno is only meaningful for a POSIX/C error (err == -1). extra = errno return (e.err, extra) @@ -400,6 +419,8 @@ def hipFileStreamRegister(uintptr_t stream_handle, unsigned flags=0): e = _c.hipFileStreamRegister(<_c.hipStream_t>stream_handle, flags) if e.err == _c.hipFileHipDriverError: extra = e.hip_drv_err + elif e.err == -1: + extra = errno # POSIX/C error return (e.err, extra) @@ -411,6 +432,8 @@ def hipFileStreamDeregister(uintptr_t stream_handle): e = _c.hipFileStreamDeregister(<_c.hipStream_t>stream_handle) if e.err == _c.hipFileHipDriverError: extra = e.hip_drv_err + elif e.err == -1: + extra = errno # POSIX/C error return (e.err, extra) @@ -423,6 +446,11 @@ def supports_async(): cdef _c.hipFileError_t e with nogil: e = _c.hipFileStreamRegister(<_c.hipStream_t>0, 0) + # If the probe actually registered the default stream, undo it so the + # probe leaves no side effect (a leaked permanent registration would + # make a later user register fail with AlreadyRegistered). + if e.err == _c.hipFileSuccess: + _c.hipFileStreamDeregister(<_c.hipStream_t>0) return e.err != _c.hipFileAsyncNotSupported diff --git a/projects/hipfile/python/hipfile/file.py b/projects/hipfile/python/hipfile/file.py index 7d30075c2be..cb882636bf0 100644 --- a/projects/hipfile/python/hipfile/file.py +++ b/projects/hipfile/python/hipfile/file.py @@ -323,9 +323,12 @@ def read_async( raise RuntimeError("The FileHandle is not open.") io = AsyncIOHandle(size, file_offset, buffer_offset) err, extra_err = hipFileReadAsync(self._handle, buffer.ptr, io, stream) + if err == -1: + # POSIX/C error: extra_err is errno (matches read()). + raise OSError(extra_err, os.strerror(extra_err)) if err != 0: # err is hipFileOpError_t; extra_err is hipError_t when err == - # hipFileHipDriverError, else errno (advisory). + # hipFileHipDriverError. raise HipFileException(err, extra_err) return io @@ -346,6 +349,9 @@ def write_async( raise RuntimeError("The FileHandle is not open.") io = AsyncIOHandle(size, file_offset, buffer_offset) err, extra_err = hipFileWriteAsync(self._handle, buffer.ptr, io, stream) + if err == -1: + # POSIX/C error: extra_err is errno (matches write()). + raise OSError(extra_err, os.strerror(extra_err)) if err != 0: raise HipFileException(err, extra_err) return io @@ -395,10 +401,10 @@ def deregister(self) -> None: """Deregister the stream. Idempotent.""" if not self._registered: return - self._registered = False err, extra = hipFileStreamDeregister(self._stream) if err != 0: raise HipFileException(err, extra) + self._registered = False def __enter__(self) -> Stream: self.register()