From 0fb192c4c32583212d582dbf125bc9c8c8db4ac3 Mon Sep 17 00:00:00 2001 From: clem-gb Date: Thu, 8 Oct 2026 09:34:47 +0200 Subject: [PATCH] fix(libkernel): use positioned transfers for Windows AIO requests On Windows each AIO request seeked a _dup of the descriptor and then read or wrote through it. A duplicated descriptor shares the file position, and the seek and the transfer are separate calls, so two threads submitting requests on one descriptor could transfer at each other's offset. Each request is now a single ReadFile/WriteFile with an OVERLAPPED offset on the OS handle, as pread and pwrite already do in Stdio.cpp. GuestAio reads four distinct blocks of one descriptor from four threads and checks every result. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01HeC9bA3n74TzJRYv9AgBE8 --- core/libs/prx/libkernel/Aio/src/Aio.cpp | 65 ++++++++++++------------- core/libs/tests/GuestAio.cpp | 28 +++++++++++ 2 files changed, 60 insertions(+), 33 deletions(-) diff --git a/core/libs/prx/libkernel/Aio/src/Aio.cpp b/core/libs/prx/libkernel/Aio/src/Aio.cpp index 0dc7d97d8..bee34799c 100644 --- a/core/libs/prx/libkernel/Aio/src/Aio.cpp +++ b/core/libs/prx/libkernel/Aio/src/Aio.cpp @@ -13,9 +13,9 @@ #include "prx/libkernel/KernelErrors.hpp" #ifdef _WIN32 -#include #include #include +#include #else #include #include @@ -54,6 +54,35 @@ void SetState(std::int32_t id, std::int32_t state) { g_states[id] = state; } +#ifdef _WIN32 +std::int64_t NativePositioned(std::int32_t fd, void* buf, std::size_t nbyte, std::int64_t offset, bool write, const char* name) { + if (nbyte > static_cast(std::numeric_limits::max())) { + throw std::runtime_error(std::string(name) + ": nbytes exceeds platform limit"); + } + if (offset < 0) { + errno = EINVAL; + return -1; + } + const auto handle = reinterpret_cast(::_get_osfhandle(fd)); + if (handle == INVALID_HANDLE_VALUE) { + errno = EBADF; + return -1; + } + OVERLAPPED overlapped{}; + overlapped.Offset = static_cast(offset); + overlapped.OffsetHigh = static_cast(static_cast(offset) >> 32u); + DWORD done = 0; + const BOOL ok = write ? ::WriteFile(handle, buf, static_cast(nbyte), &done, &overlapped) + : ::ReadFile(handle, buf, static_cast(nbyte), &done, &overlapped); + if (!ok) { + if (!write && ::GetLastError() == ERROR_HANDLE_EOF) return 0; + errno = EIO; + return -1; + } + return done; +} +#endif + std::int64_t NativePread(std::int32_t fd, void* buf, std::size_t nbyte, std::int64_t offset) { const GuestArena::HostWrite destination(buf, nbyte); if (!destination.Open()) { @@ -61,22 +90,7 @@ std::int64_t NativePread(std::int32_t fd, void* buf, std::size_t nbyte, std::int return -1; } #ifdef _WIN32 - if (nbyte > static_cast(std::numeric_limits::max())) { - throw std::runtime_error("sceKernelAioSubmitReadCommands: nbytes exceeds platform limit"); - } - const int duped = ::_dup(fd); - if (duped < 0) return -1; - if (::_lseeki64(duped, offset, SEEK_SET) < 0) { - const int error = errno; - ::_close(duped); - errno = error; - return -1; - } - const int result = ::_read(duped, buf, static_cast(nbyte)); - const int error = errno; - ::_close(duped); - errno = error; - return result; + return NativePositioned(fd, buf, nbyte, offset, false, "sceKernelAioSubmitReadCommands"); #else return static_cast(::pread(fd, buf, nbyte, static_cast(offset))); #endif @@ -84,22 +98,7 @@ std::int64_t NativePread(std::int32_t fd, void* buf, std::size_t nbyte, std::int std::int64_t NativePwrite(std::int32_t fd, const void* buf, std::size_t nbyte, std::int64_t offset) { #ifdef _WIN32 - if (nbyte > static_cast(std::numeric_limits::max())) { - throw std::runtime_error("sceKernelAioSubmitWriteCommands: nbytes exceeds platform limit"); - } - const int duped = ::_dup(fd); - if (duped < 0) return -1; - if (::_lseeki64(duped, offset, SEEK_SET) < 0) { - const int error = errno; - ::_close(duped); - errno = error; - return -1; - } - const int result = ::_write(duped, buf, static_cast(nbyte)); - const int error = errno; - ::_close(duped); - errno = error; - return result; + return NativePositioned(fd, const_cast(buf), nbyte, offset, true, "sceKernelAioSubmitWriteCommands"); #else return static_cast(::pwrite(fd, buf, nbyte, static_cast(offset))); #endif diff --git a/core/libs/tests/GuestAio.cpp b/core/libs/tests/GuestAio.cpp index 7193da905..84d3f22cd 100644 --- a/core/libs/tests/GuestAio.cpp +++ b/core/libs/tests/GuestAio.cpp @@ -3,6 +3,7 @@ #include "prx/libkernel/File/include/FileFlags.hpp" #include "prx/libkernel/KernelErrors.hpp" #include +#include #include #include #include @@ -11,6 +12,8 @@ #include #include #include +#include +#include extern "C" { int APS5_VABI sceKernelOpen(const char* path, int flags, std::uint16_t mode); int APS5_VABI sceKernelClose(int d); @@ -178,5 +181,30 @@ int main() { Check(sceKernelAioDeleteRequests(badIds, 2, untouched) == SCE_KERNEL_ERROR_EINVAL); Check(untouched[0] == 7 && untouched[1] == 7); Check(sceKernelAioDeleteRequests(writeIds, 2, nullptr) == SCE_KERNEL_ERROR_EFAULT); + const auto shared = root / "shared.bin"; + std::string sharedContents; + for (char letter : {'A', 'B', 'C', 'D'}) sharedContents.append(1024, letter); + { std::ofstream stream(shared, std::ios::binary); stream << sharedContents; } + const int sharedFd = sceKernelOpen(shared.string().c_str(), SCE_KERNEL_O_RDONLY, 0); + Check(sharedFd >= 0); + std::atomic misread{false}; + std::vector readers; + for (std::int64_t offset = 0; offset < 4096; offset += 1024) { + readers.emplace_back([&, offset] { + for (int i = 0; i < 2000 && !misread; ++i) { + std::array bytes{}; + KernelAioResult result{-1, 0}; + KernelAioRwRequest request{offset, bytes.size(), bytes.data(), &result, sharedFd}; + std::int32_t requestId = 0; + if (sceKernelAioSubmitReadCommands(&request, 1, 0, &requestId) != 0 || result.state != 3 || + result.return_value != 16 || std::memcmp(bytes.data(), sharedContents.data() + offset, 16) != 0) { + misread = true; + } + } + }); + } + for (auto& reader : readers) reader.join(); + Check(!misread); + Check(sceKernelClose(sharedFd) == 0); Check(std::filesystem::remove_all(root) > 0); }