diff --git a/python/tests/test_typing.py b/python/tests/test_typing.py index d566d5efc..cf326699b 100644 --- a/python/tests/test_typing.py +++ b/python/tests/test_typing.py @@ -112,7 +112,7 @@ def test_index_type_has_member(member): assert member in IndexType.__members__ -@pytest.mark.parametrize("member", ["PREAD", "LIBAIO"]) +@pytest.mark.parametrize("member", ["PREAD", "LIBAIO", "IO_URING"]) def test_io_backend_type_has_member(member): assert member in IOBackendType.__members__ diff --git a/python/zvec/__init__.pyi b/python/zvec/__init__.pyi index 09e828dae..1faadebd4 100644 --- a/python/zvec/__init__.pyi +++ b/python/zvec/__init__.pyi @@ -52,7 +52,9 @@ from .zvec import create_and_open, init, open def io_backend_type() -> IOBackendType: """Returns the current I/O backend type for DiskAnn async disk reads as an IOBackendType enum (zvec.typing.IOBackendType). - IOBackendType.LIBAIO if libaio is available, IOBackendType.PREAD otherwise.""" + IOBackendType.IO_URING if io_uring is available, + IOBackendType.LIBAIO if libaio is available, + IOBackendType.PREAD otherwise.""" def io_backend_description() -> str: """Returns a human-readable description of the current I/O backend. diff --git a/python/zvec/typing/__init__.pyi b/python/zvec/typing/__init__.pyi index fab03f4b9..4611955bd 100644 --- a/python/zvec/typing/__init__.pyi +++ b/python/zvec/typing/__init__.pyi @@ -130,6 +130,7 @@ class IOBackendType: - PREAD: Synchronous pread() — no async I/O. - LIBAIO: libaio loaded at runtime via dlopen(). + - IO_URING: io_uring via raw kernel syscalls (zero dependency). Examples: >>> from zvec.typing import IOBackendType @@ -142,13 +143,16 @@ class IOBackendType: PREAD LIBAIO + + IO_URING """ + IO_URING: typing.ClassVar[IOBackendType] # value = LIBAIO: typing.ClassVar[IOBackendType] # value = PREAD: typing.ClassVar[IOBackendType] # value = __members__: typing.ClassVar[ dict[str, IOBackendType] - ] # value = {'PREAD': , 'LIBAIO': } + ] # value = {'PREAD': , 'LIBAIO': , 'IO_URING': } def __eq__(self, other: typing.Any) -> bool: ... def __getstate__(self) -> int: ... diff --git a/src/ailego/io/io_backend_def.h b/src/ailego/io/io_backend_def.h index d795c3046..9cf8034a0 100644 --- a/src/ailego/io/io_backend_def.h +++ b/src/ailego/io/io_backend_def.h @@ -12,32 +12,36 @@ // See the License for the specific language governing permissions and // limitations under the License. -// Abstract I/O backend selector. +// Abstract I/O backend selector — internal header. // -// Wraps the low-level loaders (LibAioLoader for libaio) and provides a uniform -// way to initialize, query, and report the active I/O backend. The actual I/O -// operations are still performed by the underlying loaders; this class is -// responsible only for backend initialization and reporting. +// Wraps the low-level backends (io_uring via raw syscalls, LibAioLoader for +// libaio) and provides a uniform way to initialize, query, and report the +// active I/O backend. The actual I/O operations are still performed by the +// underlying backends; this class is responsible only for backend +// initialization and reporting. // // When no async backend is available, the caller should fall back to // synchronous pread(). -// -// Usage: -// auto& backend = ailego::IOBackend::Instance(); -// if (!backend.is_pread()) { ... } -// LOG_INFO("I/O backend: %s", backend.name()); #pragma once #include #include +#if defined(__linux) || defined(__linux__) +#include // ::syscall(), ::close() — POSIX only +#include // std::memset +#include // io_uring_params, __NR_io_uring_setup +#endif + namespace zvec { namespace ailego { // Returns a human-readable name for the given backend type. inline const char *IOBackendTypeName(IOBackendType type) { switch (type) { + case IOBackendType::kIoUring: + return "io_uring"; case IOBackendType::kLibAio: return "libaio"; case IOBackendType::kPread: @@ -50,6 +54,9 @@ inline const char *IOBackendTypeName(IOBackendType type) { // When the backend is kPread, includes installation guidance for libaio. inline const char *IOBackendDescription(IOBackendType type) { switch (type) { + case IOBackendType::kIoUring: + return "io_uring async I/O backend (raw kernel syscalls, zero " + "dependency)."; case IOBackendType::kLibAio: return "libaio async I/O backend loaded at runtime via dlopen()."; case IOBackendType::kPread: @@ -63,8 +70,8 @@ inline const char *IOBackendDescription(IOBackendType type) { // Singleton that loads and queries an I/O backend on demand. // -// available() (no arg) tries the best backend with priority (libaio > pread) -// and returns the loaded backend type. +// available() (no arg) tries the best backend with priority +// (io_uring > libaio > pread) and returns the loaded backend type. // available(IOBackendType) tries a specific backend. // Use type() / name() to query the loaded backend without triggering a load. class IOBackend { @@ -74,14 +81,18 @@ class IOBackend { return instance; } - // Try to load the best available backend (libaio > pread). + // Try to load the best available backend (io_uring > libaio > pread). // Returns the loaded backend type. // Idempotent — if already loaded, returns immediately. IOBackendType available() { if (type_ != IOBackendType::kPread) { return type_; } - return available(IOBackendType::kLibAio); + IOBackendType t = available(IOBackendType::kIoUring); + if (t == IOBackendType::kPread) { + t = available(IOBackendType::kLibAio); + } + return t; } // Try to load the requested backend. Returns the loaded backend type @@ -92,6 +103,18 @@ class IOBackend { return type_; } #if defined(__linux) || defined(__linux__) + if (requested == IOBackendType::kIoUring) { + // Probe io_uring availability with a minimal ring setup using only + // raw syscalls — no dependency on liburing. + struct io_uring_params params; + std::memset(¶ms, 0, sizeof(params)); + int fd = static_cast(::syscall(__NR_io_uring_setup, 1, ¶ms)); + if (fd >= 0) { + ::close(fd); + type_ = IOBackendType::kIoUring; + return type_; + } + } if (requested == IOBackendType::kLibAio) { if (LibAioLoader::Instance().load() && LibAioLoader::Instance().is_available()) { @@ -112,6 +135,10 @@ class IOBackend { return available() == IOBackendType::kLibAio; } + bool is_io_uring() { + return available() == IOBackendType::kIoUring; + } + // Returns the loaded backend type. IOBackendType type() const { return type_; diff --git a/src/ailego/io/iouring_def.h b/src/ailego/io/iouring_def.h new file mode 100644 index 000000000..ce87b246e --- /dev/null +++ b/src/ailego/io/iouring_def.h @@ -0,0 +1,199 @@ +// Copyright 2025-present the zvec project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Private header defining the io_uring kernel ABI structures and constants. +// +// This header is the io_uring analogue of libaio_def.h: it declares *only* +// the types, constants, and inline helpers that zvec needs from the io_uring +// kernel interface. By defining these structures ourselves we avoid any +// build-time dependency on or liburing-dev, mirroring the +// project's zero-dependency philosophy established by the libaio dlopen +// approach. +// +// The struct layouts (io_uring_sqe, io_uring_cqe, io_uring_params, +// io_sqring_offsets, io_cqring_offsets) are part of the Linux kernel ABI +// and are copied verbatim from . + +#pragma once + +#include + +#if defined(__linux) || defined(__linux__) + +// --------------------------------------------------------------------------- +// Syscall numbers +// --------------------------------------------------------------------------- +// io_uring was introduced in Linux 5.1 (2019). The three syscalls share the +// same numbers across all supported architectures. We prefer the values +// from when available and fall back to hardcoded numbers. +#include + +#ifndef __NR_io_uring_setup +#define __NR_io_uring_setup 425 +#endif + +#ifndef __NR_io_uring_enter +#define __NR_io_uring_enter 426 +#endif + +#ifndef __NR_io_uring_register +#define __NR_io_uring_register 427 +#endif + +// --------------------------------------------------------------------------- +// Constants (from ) +// --------------------------------------------------------------------------- + +// mmap offsets for the three shared regions. +#define IORING_OFF_SQ_RING 0ULL +#define IORING_OFF_CQ_RING 0x8000000ULL +#define IORING_OFF_SQES 0x10000000ULL + +// io_uring_enter flags. +#define IORING_ENTER_GETEVENTS (1U << 0) + +// io_uring_setup flags (none used by default). +// IORING_SETUP_IOPOLL (1U << 0) +// IORING_SETUP_SQPOLL (1U << 1) +// IORING_SETUP_SQ_AFF (1U << 2) + +// SQE opcode values. +#define IORING_OP_NOP 0 +#define IORING_OP_READV 1 +#define IORING_OP_WRITEV 2 +#define IORING_OP_FSYNC 3 +#define IORING_OP_READ_FIXED 4 +#define IORING_OP_WRITE_FIXED 5 +#define IORING_OP_POLL_ADD 6 +#define IORING_OP_POLL_REMOVE 7 +#define IORING_OP_SYNC_FILE_RANGE 8 +#define IORING_OP_SENDMSG 9 +#define IORING_OP_RECVMSG 10 +#define IORING_OP_TIMEOUT 11 +#define IORING_OP_TIMEOUT_REMOVE 12 +#define IORING_OP_ACCEPT 13 +#define IORING_OP_ASYNC_CANCEL 14 +#define IORING_OP_LINK_TIMEOUT 15 +#define IORING_OP_CONNECT 16 +#define IORING_OP_FALLOCATE 17 +#define IORING_OP_OPENAT 18 +#define IORING_OP_CLOSE 19 +#define IORING_OP_FILES_UPDATE 20 +#define IORING_OP_STATX 21 +#define IORING_OP_READ 22 +#define IORING_OP_WRITE 23 + +// --------------------------------------------------------------------------- +// Struct definitions (copied verbatim from ) +// --------------------------------------------------------------------------- + +// Submission queue entry — 64 bytes. +struct io_uring_sqe { + uint8_t opcode; // type of operation for this sqe + uint8_t flags; // IOSQE_ flags + uint16_t ioprio; // ioprio for the request + int32_t fd; // file descriptor to do IO on + union { + uint64_t off; // offset into file + uint64_t addr2; + }; + union { + uint64_t addr; // buffer or iovecs + uint64_t splice_off_in; + }; + uint32_t len; // buffer size or number of iovecs + union { + uint32_t rw_flags; // read/write flags (union of all flag types) + }; + uint64_t user_data; // data to be passed back at completion time + union { + struct { + uint16_t buf_index; // index into fixed buffers, if used + uint16_t personality; + } buf; + uint64_t __pad2[3]; + }; +}; + +// Completion queue entry — 16 bytes. +struct io_uring_cqe { + uint64_t user_data; // sqe->user_data + int32_t res; // result code for this event + uint32_t flags; +}; + +// SQ ring offsets — returned by io_uring_setup in io_uring_params. +struct io_sqring_offsets { + uint32_t head; + uint32_t tail; + uint32_t ring_mask; + uint32_t ring_entries; + uint32_t flags; + uint32_t dropped; + uint32_t array; + uint32_t resv1; + uint64_t resv2; +}; + +// CQ ring offsets — returned by io_uring_setup in io_uring_params. +struct io_cqring_offsets { + uint32_t head; + uint32_t tail; + uint32_t ring_mask; + uint32_t ring_entries; + uint32_t overflow; + uint32_t cqes; + uint32_t flags; + uint32_t resv1; + uint64_t resv2; +}; + +// Parameters passed to io_uring_setup(). +struct io_uring_params { + uint32_t sq_entries; + uint32_t cq_entries; + uint32_t flags; + uint32_t sq_thread_cpu; + uint32_t sq_thread_idle; + uint32_t features; + uint32_t wq_fd; + uint32_t resv[3]; + struct io_sqring_offsets sq_off; + struct io_cqring_offsets cq_off; +}; + +// --------------------------------------------------------------------------- +// Inline helper — prepare an SQE for a read operation. +// --------------------------------------------------------------------------- +static inline void io_uring_prep_read(struct io_uring_sqe *sqe, int fd, + void *buf, uint32_t nbytes, + uint64_t offset) { + sqe->opcode = IORING_OP_READ; + sqe->flags = 0; + sqe->ioprio = 0; + sqe->fd = fd; + sqe->off = offset; + sqe->addr = reinterpret_cast(buf); + sqe->len = nbytes; + sqe->rw_flags = 0; + sqe->user_data = 0; + sqe->buf.buf_index = 0; + sqe->buf.personality = 0; +} + +// --------------------------------------------------------------------------- +// End: struct and constant definitions from +// --------------------------------------------------------------------------- + +#endif // __linux__ diff --git a/src/ailego/io/iouring_loader.h b/src/ailego/io/iouring_loader.h new file mode 100644 index 000000000..0b6143fd7 --- /dev/null +++ b/src/ailego/io/iouring_loader.h @@ -0,0 +1,313 @@ +// Copyright 2025-present the zvec project +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +// Raw-syscall wrapper for Linux io_uring. +// +// This class implements the io_uring submission/completion queue lifecycle +// using *only* kernel syscalls (io_uring_setup, io_uring_enter) and mmap. +// There is zero dependency on liburing or liburing-dev: no build-time header, +// no runtime .so, and no dlopen. This mirrors the project's existing +// zero-dependency philosophy established by the libaio dlopen approach, but +// goes one step further — io_uring is a pure kernel ABI accessed via syscall. +// +// Runtime detection is automatic: if the kernel does not support io_uring +// (pre-5.1 or io_uring disabled), io_uring_setup() returns -ENOSYS and +// setup() returns false, allowing callers to fall back to libaio or pread. +// +// The ring is **not** thread-safe. Each thread that performs I/O must have +// its own IoUringRing instance (managed through IoBackend / IOContext). + +#pragma once + +#if defined(__linux) || defined(__linux__) + +#include +#include +#include +#include +#include +#include +#include +#include + +namespace zvec { +namespace core { + +// Forward declaration — AlignedRead is defined in diskann_file_reader.h +// after this header is included. The forward declaration is sufficient +// because IoUringRing::execute takes it by reference. +struct AlignedRead; + +// Maximum number of SQEs we submit in a single io_uring_enter() call. +static constexpr uint32_t kIoUringMaxBatch = 128; + +// Alignment of every staging slot — keeps O_DIRECT reads legal. +static constexpr size_t kIoUringStagingAlign = 4096; + +class IoUringRing { + public: + IoUringRing() = default; + ~IoUringRing() { + teardown(); + } + + IoUringRing(const IoUringRing &) = delete; + IoUringRing &operator=(const IoUringRing &) = delete; + + // Create an io_uring with the given number of entries. + // Returns true on success, false if the kernel does not support io_uring + // or setup failed for any reason. + bool setup(uint32_t entries) { + struct io_uring_params params; + std::memset(¶ms, 0, sizeof(params)); + + // io_uring_setup is a raw syscall — returns fd (>=0) or -1 with errno. + ring_fd_ = static_cast( + syscall(__NR_io_uring_setup, static_cast(entries), ¶ms)); + if (ring_fd_ < 0) { + // ENOSYS = kernel doesn't support io_uring. + // EPERM = io_uring disabled via sysctl. + // EINVAL = invalid parameters. + if (errno != ENOSYS) { + LOG_WARN("io_uring_setup failed; errno=%d, %s", errno, + ::strerror(errno)); + } + return false; + } + + sq_entries_ = params.sq_entries; + cq_entries_ = params.cq_entries; + + // --- mmap the three shared regions --- + + // 1. SQ ring (includes head, tail, mask, entries, flags, dropped, array). + size_t sq_ring_sz = static_cast(params.sq_off.array) + + sq_entries_ * sizeof(uint32_t); + sq_ring_ptr_ = ::mmap(nullptr, sq_ring_sz, PROT_READ | PROT_WRITE, + MAP_SHARED, ring_fd_, IORING_OFF_SQ_RING); + if (sq_ring_ptr_ == MAP_FAILED) { + LOG_ERROR("mmap SQ ring failed: %s", ::strerror(errno)); + sq_ring_ptr_ = nullptr; + teardown(); + return false; + } + + // 2. SQE array. + size_t sqes_sz = sq_entries_ * sizeof(struct io_uring_sqe); + sqes_ptr_ = reinterpret_cast( + ::mmap(nullptr, sqes_sz, PROT_READ | PROT_WRITE, MAP_SHARED, ring_fd_, + IORING_OFF_SQES)); + if (sqes_ptr_ == MAP_FAILED) { + LOG_ERROR("mmap SQEs failed: %s", ::strerror(errno)); + sqes_ptr_ = nullptr; + teardown(); + return false; + } + + // 3. CQ ring (includes head, tail, mask, entries, overflow, cqes[]). + size_t cq_ring_sz = static_cast(params.cq_off.cqes) + + cq_entries_ * sizeof(struct io_uring_cqe); + cq_ring_ptr_ = ::mmap(nullptr, cq_ring_sz, PROT_READ | PROT_WRITE, + MAP_SHARED, ring_fd_, IORING_OFF_CQ_RING); + if (cq_ring_ptr_ == MAP_FAILED) { + LOG_ERROR("mmap CQ ring failed: %s", ::strerror(errno)); + cq_ring_ptr_ = nullptr; + teardown(); + return false; + } + + // --- Set up typed pointers into the mmap'd regions --- + + // SQ ring fields. + sq_head_ = reinterpret_cast(static_cast(sq_ring_ptr_) + + params.sq_off.head); + sq_tail_ = reinterpret_cast(static_cast(sq_ring_ptr_) + + params.sq_off.tail); + sq_ring_mask_ = reinterpret_cast( + static_cast(sq_ring_ptr_) + params.sq_off.ring_mask); + sq_ring_entries_ = reinterpret_cast( + static_cast(sq_ring_ptr_) + params.sq_off.ring_entries); + sq_flags_ = reinterpret_cast(static_cast(sq_ring_ptr_) + + params.sq_off.flags); + sq_dropped_ = reinterpret_cast( + static_cast(sq_ring_ptr_) + params.sq_off.dropped); + sq_array_ = reinterpret_cast(static_cast(sq_ring_ptr_) + + params.sq_off.array); + + // CQ ring fields. + cq_head_ = reinterpret_cast(static_cast(cq_ring_ptr_) + + params.cq_off.head); + cq_tail_ = reinterpret_cast(static_cast(cq_ring_ptr_) + + params.cq_off.tail); + cq_ring_mask_ = reinterpret_cast( + static_cast(cq_ring_ptr_) + params.cq_off.ring_mask); + cq_ring_entries_ = reinterpret_cast( + static_cast(cq_ring_ptr_) + params.cq_off.ring_entries); + cq_overflow_ = reinterpret_cast( + static_cast(cq_ring_ptr_) + params.cq_off.overflow); + cqes_ = reinterpret_cast( + static_cast(cq_ring_ptr_) + params.cq_off.cqes); + + // SQE array. + sqes_ = sqes_ptr_; + + // Initialize the SQ array to identity mapping so that logical index == + // physical SQE index. This is the simplest and most common configuration. + for (uint32_t i = 0; i < sq_entries_; i++) { + sq_array_[i] = i; + } + + return true; + } + + // Tear down the ring: munmap all regions and close the ring fd. + // + // Closing the ring fd only *initiates* asynchronous cancellation of any + // in-flight request (the kernel's release path defers the real cleanup), + // so this must only be called when the ring is quiesced — or after + // abandon_staging() has leaked the staging pool the kernel writes into. + void teardown() { + if (sq_ring_ptr_ && sq_ring_ptr_ != MAP_FAILED) { + // We don't track the exact mmap size; munmap with a large enough size + // is safe because the kernel only unmaps what was actually mapped. + // However, to be correct we use the page-aligned size. + size_t sz = static_cast(sq_entries_) * sizeof(uint32_t) + 4096; + ::munmap(sq_ring_ptr_, sz); + } + if (sqes_ptr_ && sqes_ptr_ != MAP_FAILED) { + size_t sz = + static_cast(sq_entries_) * sizeof(struct io_uring_sqe); + ::munmap(sqes_ptr_, sz); + } + if (cq_ring_ptr_ && cq_ring_ptr_ != MAP_FAILED) { + size_t sz = + static_cast(cq_entries_) * sizeof(struct io_uring_cqe) + 4096; + ::munmap(cq_ring_ptr_, sz); + } + + sq_ring_ptr_ = nullptr; + sqes_ptr_ = nullptr; + cq_ring_ptr_ = nullptr; + sqes_ = nullptr; + cqes_ = nullptr; + sq_head_ = sq_tail_ = sq_ring_mask_ = sq_ring_entries_ = nullptr; + sq_flags_ = sq_dropped_ = sq_array_ = nullptr; + cq_head_ = cq_tail_ = cq_ring_mask_ = cq_ring_entries_ = nullptr; + cq_overflow_ = nullptr; + + if (ring_fd_ >= 0) { + ::close(ring_fd_); + ring_fd_ = -1; + } + + ::free(staging_); + staging_ = nullptr; + staging_size_ = 0; + + sq_entries_ = 0; + cq_entries_ = 0; + } + + // Grow the staging pool so a whole batch of reads can land in ring-owned + // memory. Only safe to call while the ring is quiesced, because the + // previous pool is freed here and must no longer be referenced by the + // kernel. + bool ensure_staging(size_t bytes) { + if (staging_size_ >= bytes) { + return true; + } + void *ptr = nullptr; + int err = ::posix_memalign(&ptr, kIoUringStagingAlign, bytes); + if (err != 0) { + LOG_WARN("io_uring staging allocation failed; size=%zu, %s", bytes, + ::strerror(err)); + return false; + } + ::free(staging_); + staging_ = static_cast(ptr); + staging_size_ = bytes; + return true; + } + + // Deliberately leak the staging pool. Called when in-flight requests + // could not be drained: the kernel may keep writing into the pool even + // after the ring fd is closed (io_uring teardown is asynchronous), so + // the memory must outlive this object. The caller's destination buffers + // are never handed to the kernel, so they remain safe to reuse or free. + void abandon_staging() { + staging_ = nullptr; + staging_size_ = 0; + } + + bool is_valid() const { + return ring_fd_ >= 0; + } + + // Execute a batch of aligned read requests via io_uring. + // + // Implemented in diskann_file_reader.cc to avoid a circular dependency: + // AlignedRead is defined in diskann_file_reader.h *after* this header is + // included, so the method body cannot be inline here. + // + // On success returns 0. On failure returns -1; the caller may always + // fall back to pread, because the kernel only ever writes into the + // ring-owned staging pool — completed reads are copied into the caller's + // buffers, which are never exposed to in-flight requests. If the ring + // cannot be drained, the staging pool is leaked and the ring torn down. + int execute(int fd, std::vector &read_reqs); + + private: + int ring_fd_{-1}; + + // mmap'd region base pointers (needed for munmap). + void *sq_ring_ptr_{nullptr}; + struct io_uring_sqe *sqes_ptr_{nullptr}; + void *cq_ring_ptr_{nullptr}; + + // SQ ring field pointers (into sq_ring_ptr_). + unsigned *sq_head_{nullptr}; + unsigned *sq_tail_{nullptr}; + unsigned *sq_ring_mask_{nullptr}; + unsigned *sq_ring_entries_{nullptr}; + unsigned *sq_flags_{nullptr}; + unsigned *sq_dropped_{nullptr}; + unsigned *sq_array_{nullptr}; + + // CQ ring field pointers (into cq_ring_ptr_). + unsigned *cq_head_{nullptr}; + unsigned *cq_tail_{nullptr}; + unsigned *cq_ring_mask_{nullptr}; + unsigned *cq_ring_entries_{nullptr}; + unsigned *cq_overflow_{nullptr}; + struct io_uring_cqe *cqes_{nullptr}; + + // SQE array pointer. + struct io_uring_sqe *sqes_{nullptr}; + + // Ring-owned staging pool. The kernel reads into this pool instead of + // the caller's buffers; execute() copies verified completions out. This + // decouples the caller's buffer lifetime from asynchronous io_uring + // teardown (see abandon_staging()). + char *staging_{nullptr}; + size_t staging_size_{0}; + + // Ring capacities. + unsigned sq_entries_{0}; + unsigned cq_entries_{0}; +}; + +} // namespace core +} // namespace zvec + +#endif // __linux__ diff --git a/src/binding/python/model/common/python_config.cc b/src/binding/python/model/common/python_config.cc index a9305dac9..ced18b066 100644 --- a/src/binding/python/model/common/python_config.cc +++ b/src/binding/python/model/common/python_config.cc @@ -228,6 +228,7 @@ void ZVecPyConfig::Initialize(pybind11::module_ &m) { }, "Returns the current I/O backend type for DiskAnn async disk reads " "as an IOBackendType enum (zvec.typing.IOBackendType). " + "IOBackendType.IO_URING if io_uring is available, " "IOBackendType.LIBAIO if libaio is available, " "IOBackendType.PREAD otherwise."); diff --git a/src/binding/python/typing/python_type.cc b/src/binding/python/typing/python_type.cc index d20cac755..45d3d8216 100644 --- a/src/binding/python/typing/python_type.cc +++ b/src/binding/python/typing/python_type.cc @@ -146,6 +146,7 @@ Enumeration of supported I/O backend types for DiskAnn async disk reads. - PREAD: Synchronous pread() \u2014 no async I/O. - LIBAIO: libaio loaded at runtime via dlopen(). +- IO_URING: io_uring via raw kernel syscalls (zero dependency). Examples: >>> from zvec.typing import IOBackendType @@ -153,7 +154,8 @@ Enumeration of supported I/O backend types for DiskAnn async disk reads. IOBackendType.LIBAIO )pbdoc") .value("PREAD", ailego::IOBackendType::kPread) - .value("LIBAIO", ailego::IOBackendType::kLibAio); + .value("LIBAIO", ailego::IOBackendType::kLibAio) + .value("IO_URING", ailego::IOBackendType::kIoUring); } void ZVecPyTyping::bind_status(py::module_ &m) { diff --git a/src/core/algorithm/diskann/diskann_file_reader.cc b/src/core/algorithm/diskann/diskann_file_reader.cc index cde0755e8..ba8221510 100644 --- a/src/core/algorithm/diskann/diskann_file_reader.cc +++ b/src/core/algorithm/diskann/diskann_file_reader.cc @@ -16,9 +16,11 @@ #include #include #include +#include #include #include #include +#include #include #include #include @@ -32,6 +34,10 @@ namespace core { typedef struct io_event io_event_t; typedef struct iocb iocb_t; +// Retry budget for draining in-flight io_uring requests when the kernel +// keeps returning EAGAIN/EBUSY (100 us sleep per retry, ~1 s total). +static constexpr size_t kIoUringDrainRetries = 10000; + // Ensures the I/O backend selection is logged exactly once per process, // regardless of which entry point (setup_io_ctx or register_thread) // triggers it first. @@ -58,11 +64,37 @@ int setup_io_ctx(IOContext &ctx) { #if (defined(__linux) || defined(__linux__)) std::call_once(g_io_backend_log_once, log_diskann_io_backend); if (ailego::IOBackend::Instance().is_pread()) { + // No async backend available — leave ctx null so callers fall back to + // synchronous pread(). return 0; } - int ret = LibAioLoader::Instance().io_setup(MAX_EVENTS, &ctx); - return ret; + ctx = new IoBackend(); + ailego::IOBackendType selected = ailego::IOBackend::Instance().available(); + + // Priority 1: io_uring (raw kernel syscalls — zero dependency). + if (selected == ailego::IOBackendType::kIoUring && + ctx->ring.setup(MAX_EVENTS)) { + ctx->backend = IoBackend::IO_URING; + return 0; + } + + // Priority 2: libaio (dlopen — soft dependency). + if (selected != ailego::IOBackendType::kPread && + LibAioLoader::Instance().load() && + LibAioLoader::Instance().is_available()) { + int ret = LibAioLoader::Instance().io_setup(MAX_EVENTS, &ctx->aio_ctx); + if (ret == 0) { + ctx->backend = IoBackend::LIBAIO; + return 0; + } + LOG_WARN("io_setup failed; returned: %d, %s. falling back to pread", ret, + ::strerror(-ret)); + } + + // Priority 3: synchronous pread (always available). + ctx->backend = IoBackend::NONE; + return 0; #else return 0; #endif @@ -70,15 +102,21 @@ int setup_io_ctx(IOContext &ctx) { int destroy_io_ctx(IOContext &ctx) { #if (defined(__linux) || defined(__linux__)) - if (ailego::IOBackend::Instance().is_pread() || ctx == nullptr) { + if (ctx == nullptr) { return 0; } - int ret = LibAioLoader::Instance().io_destroy(ctx); - if (ret == 0) { - ctx = nullptr; + + if (ctx->backend == IoBackend::IO_URING) { + ctx->ring.teardown(); + } else if (ctx->backend == IoBackend::LIBAIO && + LibAioLoader::Instance().is_available()) { + LibAioLoader::Instance().io_destroy(ctx->aio_ctx); } + // IoUringRing destructor also calls teardown() — idempotent and safe. - return ret; + delete ctx; + ctx = nullptr; + return 0; #else return 0; #endif @@ -107,7 +145,7 @@ static int execute_io_pread(int fd, std::vector &read_reqs) { // invalid arguments. If that happens after submission, io_destroy() is the // only safe way to quiesce the context before synchronous I/O touches the same // destination buffers. Recreate the context so later reads can still use AIO. -static bool reset_aio_context(IOContext &ctx) { +static bool reset_aio_context(io_context_t &ctx) { auto &loader = LibAioLoader::Instance(); int ret; do { @@ -121,7 +159,7 @@ static bool reset_aio_context(IOContext &ctx) { } ctx = nullptr; - IOContext replacement = nullptr; + io_context_t replacement = nullptr; ret = loader.io_setup(MAX_EVENTS, &replacement); if (ret != 0) { LOG_ERROR( @@ -134,7 +172,7 @@ static bool reset_aio_context(IOContext &ctx) { return true; } -int execute_io_libaio(IOContext &ctx, int fd, +int execute_io_libaio(io_context_t &ctx, int fd, std::vector &read_reqs, uint64_t n_retries) { uint64_t iters = DiskAnnUtil::div_round_up(read_reqs.size(), MAX_EVENTS); @@ -254,18 +292,250 @@ int execute_io_libaio(IOContext &ctx, int fd, } #endif -int execute_io(IOContext &ctx, int fd, std::vector &read_reqs, +int execute_io(IOContext ctx, int fd, std::vector &read_reqs, uint64_t n_retries = 0) { #if (defined(__linux) || defined(__linux__)) - if (ailego::IOBackend::Instance().is_pread() || ctx == nullptr) { + // Guard against null or sentinel contexts. + if (ctx == nullptr || ctx == (IOContext)-1) { return execute_io_pread(fd, read_reqs); } - return execute_io_libaio(ctx, fd, read_reqs, n_retries); + + // Dispatch based on the active backend. + if (ctx->backend == IoBackend::IO_URING) { + int ret = ctx->ring.execute(fd, read_reqs); + if (ret == 0) { + return 0; + } + // The kernel only ever writes into the ring-owned staging pool, never + // into the caller's buffers, so a pread fallback can never race with + // requests that are still in flight. + LOG_WARN("io_uring execute failed; falling back to pread"); + return execute_io_pread(fd, read_reqs); + } + + if (ctx->backend == IoBackend::LIBAIO) { + return execute_io_libaio(ctx->aio_ctx, fd, read_reqs, n_retries); + } + + // NONE backend — synchronous pread. + return execute_io_pread(fd, read_reqs); #else return execute_io_pread(fd, read_reqs); #endif } +// --------------------------------------------------------------------------- +// IoUringRing::execute — defined here (not in iouring_loader.h) because it +// accesses AlignedRead members, and AlignedRead is defined in +// diskann_file_reader.h after iouring_loader.h is included. +// --------------------------------------------------------------------------- +#if (defined(__linux) || defined(__linux__)) +int IoUringRing::execute(int fd, std::vector &read_reqs) { + if (!is_valid()) { + return -1; + } + if (read_reqs.empty()) { + return 0; + } + + // Process in batches limited by the SQ ring size. + uint32_t batch_size = + std::min(sq_entries_, static_cast(kIoUringMaxBatch)); + uint64_t iters = DiskAnnUtil::div_round_up(read_reqs.size(), batch_size); + + for (uint64_t iter = 0; iter < iters; iter++) { + uint64_t n_ops = + std::min(static_cast(read_reqs.size()) - iter * batch_size, + static_cast(batch_size)); + + // --- Phase 1: Fill SQEs --- + // + // Reads land in the ring-owned staging pool, never in the caller's + // buffers. io_uring teardown is asynchronous — closing the ring fd + // only initiates cancellation — so the kernel may still write into + // request buffers after execute() has returned an error. Staging + // memory can simply be leaked in that case (abandon_staging()), while + // the caller's buffers stay safe to reuse or free. The copy-out below + // costs one sector-scale memcpy per read, negligible next to the I/O. + std::vector slot_off(n_ops); + size_t staging_bytes = 0; + for (uint64_t j = 0; j < n_ops; j++) { + slot_off[j] = staging_bytes; + size_t len = read_reqs[j + iter * batch_size].len; + // Round every slot up so each staging pointer stays O_DIRECT-legal. + staging_bytes += + (len + kIoUringStagingAlign - 1) & ~(kIoUringStagingAlign - 1); + } + // Safe: the previous batch is fully drained before we get here, so no + // in-flight request can reference the old pool being freed on growth. + if (!ensure_staging(staging_bytes)) { + return -1; // nothing submitted; pread fallback is safe + } + + unsigned tail = __atomic_load_n(sq_tail_, __ATOMIC_ACQUIRE); + unsigned mask = *sq_ring_mask_; + + for (uint64_t j = 0; j < n_ops; j++) { + unsigned idx = (tail + static_cast(j)) & mask; + unsigned sqe_idx = sq_array_[idx]; + struct io_uring_sqe *sqe = &sqes_[sqe_idx]; + + uint64_t req_idx = j + iter * batch_size; + io_uring_prep_read(sqe, fd, staging_ + slot_off[j], + static_cast(read_reqs[req_idx].len), + read_reqs[req_idx].offset); + // Store the request index so we can verify the completion. + sqe->user_data = req_idx; + } + + // Memory barrier: ensure SQE contents are visible before tail update. + __sync_synchronize(); + __atomic_store_n(sq_tail_, tail + static_cast(n_ops), + __ATOMIC_RELEASE); + + // --- Phase 2: Submit and reap completions --- + // + // io_uring_enter() returns the number of SQEs consumed, not the number + // of CQEs available. A partial submission returns before the wait + // phase, and a signal can interrupt the wait while preserving a + // positive submission count, so IORING_ENTER_GETEVENTS guarantees + // min_complete completions only when the call finishes normally. + // Completions must therefore be counted against cq_tail instead of + // assuming n_ops CQEs are ready. + uint64_t submitted = 0; + uint64_t completed = 0; + bool all_ok = true; + + // Consume every CQE the kernel has published so far and verify it. + // Completion order is unspecified, so use cqe->user_data to find the + // request instead of assuming submission order. + auto reap_available = [&]() { + unsigned chead = *cq_head_; // single consumer — plain load is enough + unsigned ctail = __atomic_load_n(cq_tail_, __ATOMIC_ACQUIRE); + unsigned cq_mask = *cq_ring_mask_; + if (chead == ctail) { + return; + } + while (chead != ctail) { + struct io_uring_cqe *cqe = &cqes_[chead & cq_mask]; + uint64_t req_idx = cqe->user_data; + + if (req_idx < iter * batch_size || + req_idx >= iter * batch_size + n_ops) { + LOG_WARN("io_uring completion referenced unknown request: %lu", + (unsigned long)req_idx); + all_ok = false; + } else if (cqe->res < 0) { + LOG_WARN("io_uring read failed: req=%lu, res=%d, offset=%lu", + (unsigned long)req_idx, cqe->res, + (unsigned long)read_reqs[req_idx].offset); + all_ok = false; + } else if (static_cast(cqe->res) != read_reqs[req_idx].len) { + LOG_WARN("io_uring short read: req=%lu, got=%d, expected=%lu", + (unsigned long)req_idx, cqe->res, + (unsigned long)read_reqs[req_idx].len); + all_ok = false; + } else { + // Verified completion — copy from staging into the caller's + // buffer. This is the only place caller memory is written. + std::memcpy(read_reqs[req_idx].buf, + staging_ + slot_off[req_idx - iter * batch_size], + read_reqs[req_idx].len); + } + chead++; + completed++; + } + // Release: CQE reads must complete before the kernel may reuse slots. + __atomic_store_n(cq_head_, chead, __ATOMIC_RELEASE); + }; + + while (completed < n_ops) { + reap_available(); + if (completed >= n_ops) { + break; + } + + unsigned to_submit = static_cast(n_ops - submitted); + int ret = static_cast(syscall( + __NR_io_uring_enter, ring_fd_, to_submit, 1u, IORING_ENTER_GETEVENTS, + static_cast(nullptr), static_cast(0))); + if (ret >= 0) { + submitted += static_cast(ret); + continue; + } + if (errno == EINTR) { + // Interrupted during submit or wait; the SQEs already consumed are + // tracked in `submitted`, so simply retry. + continue; + } + if ((errno == EAGAIN || errno == EBUSY) && completed < submitted) { + // Kernel resources are exhausted, but in-flight requests will free + // them as they complete; keep reaping and retrying. + continue; + } + + // Unrecoverable failure (or EAGAIN with nothing in flight). + LOG_WARN( + "io_uring_enter failed; errno=%d, %s, submitted=%lu/%lu, " + "completed=%lu. draining before falling back to pread", + errno, ::strerror(errno), (unsigned long)submitted, + (unsigned long)n_ops, (unsigned long)completed); + + // Un-publish the SQEs the kernel never consumed so a later batch + // cannot submit them against stale buffers. + __atomic_store_n(sq_tail_, tail + static_cast(submitted), + __ATOMIC_RELEASE); + + // Drain every in-flight request before the staging pool may be + // freed or reused by a later batch. CQEs are posted to the shared + // ring by the kernel on its own, so completions can still be reaped + // here even when io_uring_enter() keeps failing. + size_t drain_retries = 0; + while (completed < submitted) { + reap_available(); + if (completed >= submitted) { + break; + } + int wret = static_cast(syscall( + __NR_io_uring_enter, ring_fd_, 0u, 1u, IORING_ENTER_GETEVENTS, + static_cast(nullptr), static_cast(0))); + if (wret >= 0 || errno == EINTR) { + continue; + } + if ((errno == EAGAIN || errno == EBUSY) && + drain_retries++ < kIoUringDrainRetries) { + // Give in-flight requests time to complete; entering the kernel + // via the sleep also lets pending completion task-work run. + std::this_thread::sleep_for(std::chrono::microseconds(100)); + continue; + } + // The ring cannot be drained. Leak the staging pool — the kernel + // may keep writing into it through the asynchronous teardown — and + // disable io_uring for this context. The caller's buffers were + // never exposed to the kernel, so the pread fallback stays safe. + LOG_ERROR( + "io_uring drain failed; errno=%d, %s. leaking the staging pool " + "and disabling io_uring for this context", + errno, ::strerror(errno)); + abandon_staging(); + teardown(); + return -1; + } + return -1; + } + + if (!all_ok) { + // Every request completed and the staging pool is quiesced, but at + // least one read failed or was short — let the caller retry with + // pread. + return -1; + } + } + + return 0; +} +#endif // __linux__ + LinuxAlignedFileReader::LinuxAlignedFileReader(int file_desc) { this->file_desc = file_desc; } @@ -286,7 +556,7 @@ IOContext &LinuxAlignedFileReader::get_ctx() { std::unique_lock lk(ctx_mut); auto it = ctx_map.find(std::this_thread::get_id()); if (it == ctx_map.end()) { - LOG_ERROR("bad thread access; returning -1 as io_context_t"); + LOG_ERROR("bad thread access; returning invalid IOContext"); return this->bad_ctx; } else { return it->second; @@ -299,32 +569,20 @@ void LinuxAlignedFileReader::register_thread() { std::unique_lock lk(ctx_mut); if (ctx_map.find(thread_id) != ctx_map.end()) { LOG_ERROR("multiple calls to register_thread from the same thread"); - return; } IOContext ctx = nullptr; - - std::call_once(g_io_backend_log_once, log_diskann_io_backend); - if (ailego::IOBackend::Instance().is_pread()) { + int ret = setup_io_ctx(ctx); + if (ret != 0) { + LOG_ERROR("setup_io_ctx failed; returned: %d", ret); lk.unlock(); return; } - int ret = LibAioLoader::Instance().io_setup(MAX_EVENTS, &ctx); - if (ret != 0) { - if (ret == -EAGAIN) { - LOG_ERROR( - "io_setup failed with EAGAIN: Consider increasing " - "/proc/sys/fs/aio-max-nr"); - } else { - LOG_ERROR("io_setup failed; returned: %d, %s", ret, ::strerror(-ret)); - } - } else { + if (ctx != nullptr) { LOG_INFO("allocating ctx: %lu", (uint64_t)ctx); - - ctx_map[thread_id] = ctx; } - + ctx_map[thread_id] = ctx; lk.unlock(); #endif } @@ -345,11 +603,8 @@ void LinuxAlignedFileReader::deregister_thread() { ctx_map.erase(it); } - // io_destroy is a syscall; keep it outside the lock to avoid blocking others - if (ailego::IOBackend::Instance().available() != - ailego::IOBackendType::kPread) { - LibAioLoader::Instance().io_destroy(ctx); - } + // Teardown is a syscall; keep it outside the lock to avoid blocking others. + destroy_io_ctx(ctx); LOG_INFO("returned ctx from thread"); #endif } @@ -357,13 +612,8 @@ void LinuxAlignedFileReader::deregister_thread() { void LinuxAlignedFileReader::deregister_all_threads() { #if (defined(__linux) || defined(__linux__)) std::unique_lock lk(ctx_mut); - bool aio_available = ailego::IOBackend::Instance().available() != - ailego::IOBackendType::kPread; for (auto x = ctx_map.begin(); x != ctx_map.end(); x++) { - IOContext ctx = x->second; - if (aio_available) { - LibAioLoader::Instance().io_destroy(ctx); - } + destroy_io_ctx(x->second); } ctx_map.clear(); #endif diff --git a/src/core/algorithm/diskann/diskann_file_reader.h b/src/core/algorithm/diskann/diskann_file_reader.h index a1cb7c91a..7eaf41e8b 100644 --- a/src/core/algorithm/diskann/diskann_file_reader.h +++ b/src/core/algorithm/diskann/diskann_file_reader.h @@ -18,6 +18,7 @@ #include #if (defined(__linux) || defined(__linux__)) +#include // raw-syscall io_uring wrapper (IoUringRing) #include // dlopen-based libaio wrapper #endif @@ -31,7 +32,31 @@ namespace zvec { namespace core { #if (defined(__linux) || defined(__linux__)) -typedef io_context_t IOContext; + +// IoBackend holds the per-thread I/O context for whichever async backend +// was successfully initialised at setup time. The priority is: +// 1. io_uring (raw kernel syscalls — zero dependency) +// 2. libaio (dlopen — soft dependency) +// 3. pread (always available — synchronous fallback) +// +// IOContext is a *pointer* to IoBackend, which preserves the existing +// sentinel conventions: nullptr means uninitialised and (IOContext)-1 is +// the invalid-handle sentinel returned by get_ctx() for unregistered +// threads. +struct IoBackend { + enum Backend : uint8_t { + NONE = 0, // synchronous pread + IO_URING = 1, // io_uring via raw syscalls + LIBAIO = 2, // libaio via dlopen + }; + + Backend backend{NONE}; + IoUringRing ring{}; + io_context_t aio_ctx{nullptr}; +}; + +typedef IoBackend *IOContext; + #else typedef uint32_t IOContext; #endif diff --git a/src/include/zvec/ailego/io/io_backend.h b/src/include/zvec/ailego/io/io_backend.h index 008b01514..ee00bbb1c 100644 --- a/src/include/zvec/ailego/io/io_backend.h +++ b/src/include/zvec/ailego/io/io_backend.h @@ -29,13 +29,17 @@ namespace zvec { namespace ailego { // Supported I/O backend types. +// +// Numeric values are part of the C ABI (see zvec_io_backend_type_t in c_api.h): +// kPread = 0, kLibAio = 1, kIoUring = 2. enum class IOBackendType { - kPread, // Synchronous pread() — no async I/O - kLibAio, // libaio loaded at runtime via dlopen() + kPread = 0, // Synchronous pread() — no async I/O + kLibAio = 1, // libaio loaded at runtime via dlopen() + kIoUring = 2, // io_uring via raw kernel syscalls (zero dependency) }; // Returns the currently active I/O backend type. -// Triggers backend initialization on first call (libaio > pread). +// Triggers backend initialization on first call (io_uring > libaio > pread). IOBackendType current_io_backend_type(); // Returns a human-readable description of the currently active I/O backend. diff --git a/src/include/zvec/c_api.h b/src/include/zvec/c_api.h index 2d048689b..7e18886b4 100644 --- a/src/include/zvec/c_api.h +++ b/src/include/zvec/c_api.h @@ -789,6 +789,8 @@ typedef uint32_t zvec_io_backend_type_t; 0 /**< Synchronous pread() \u2014 no async I/O */ #define ZVEC_IO_BACKEND_TYPE_LIBAIO \ 1 /**< libaio loaded at runtime via dlopen() */ +#define ZVEC_IO_BACKEND_TYPE_IO_URING \ + 2 /**< io_uring via raw kernel syscalls (zero dependency) */ /** * @brief Get the current I/O backend type for DiskAnn async disk reads. @@ -796,7 +798,8 @@ typedef uint32_t zvec_io_backend_type_t; * Pure introspection \u2014 no side effects, no install hints. * * @return zvec_io_backend_type_t The loaded backend type - * (ZVEC_IO_BACKEND_TYPE_LIBAIO or ZVEC_IO_BACKEND_TYPE_PREAD). + * (ZVEC_IO_BACKEND_TYPE_IO_URING, ZVEC_IO_BACKEND_TYPE_LIBAIO, + * or ZVEC_IO_BACKEND_TYPE_PREAD). */ ZVEC_EXPORT zvec_io_backend_type_t ZVEC_CALL zvec_get_io_backend_type(void); @@ -805,7 +808,7 @@ ZVEC_EXPORT zvec_io_backend_type_t ZVEC_CALL zvec_get_io_backend_type(void); * * @param type The backend type code. * @return Thread-local string valid until the next call on this thread; - * "libaio", "pread", or "unknown". + * "io_uring", "libaio", "pread", or "unknown". */ ZVEC_EXPORT const char *ZVEC_CALL zvec_get_io_backend_type_name(zvec_io_backend_type_t type); diff --git a/tests/core/algorithm/diskann/diskann_file_reader_aio_test.cc b/tests/core/algorithm/diskann/diskann_file_reader_aio_test.cc index fdc1a396d..a416b0623 100644 --- a/tests/core/algorithm/diskann/diskann_file_reader_aio_test.cc +++ b/tests/core/algorithm/diskann/diskann_file_reader_aio_test.cc @@ -30,7 +30,7 @@ namespace zvec { namespace core { -int execute_io_libaio(IOContext &ctx, int fd, +int execute_io_libaio(io_context_t &ctx, int fd, std::vector &read_reqs, uint64_t n_retries = 0); } // namespace core @@ -192,7 +192,7 @@ TEST(DiskAnnLinuxAioTest, AccumulatesPartialSubmissionsAndCompletions) { state.submit_results = {2, 2}; state.completion_results = {1, 2, 1}; FakeAioGuard guard(&state); - IOContext ctx = reinterpret_cast(static_cast(1)); + io_context_t ctx = reinterpret_cast(static_cast(1)); // An invalid fd makes any accidental pread fallback fail the test. EXPECT_EQ(execute_io_libaio(ctx, -1, requests), 0); @@ -221,7 +221,7 @@ TEST(DiskAnnLinuxAioTest, DrainsPartialSubmissionBeforePreadFallback) { state.submit_results = {2, -EAGAIN}; state.completion_results = {1, 1}; FakeAioGuard guard(&state); - IOContext ctx = reinterpret_cast(static_cast(1)); + io_context_t ctx = reinterpret_cast(static_cast(1)); EXPECT_EQ(execute_io_libaio(ctx, file.fd(), requests), 0); EXPECT_EQ(state.submit_sizes, (std::vector{4, 2})); @@ -248,7 +248,7 @@ TEST(DiskAnnLinuxAioTest, DrainsAllCompletionsBeforePreadFallback) { state.completion_results = {1, 1, 2}; state.short_completion = 0; FakeAioGuard guard(&state); - IOContext ctx = reinterpret_cast(static_cast(1)); + io_context_t ctx = reinterpret_cast(static_cast(1)); EXPECT_EQ(execute_io_libaio(ctx, file.fd(), requests), 0); EXPECT_EQ(state.completion_sizes, (std::vector{4, 3, 2}));