shifathosn
/
AnyPS5
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01HeC9bA3n74TzJRYv9AgBE8
This commit is contained in:
clem-gb 2026-10-08 09:34:47 +02:00
parent a5df1a874d
commit 0fb192c4c3
2 changed files with 60 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
core/libs/prx/libkernel/Aio/src/Aio.cpp
View File

@ -13,9 +13,9 @@
#include "prx/libkernel/KernelErrors.hpp"
#ifdef _WIN32
#include <fcntl.h>
#include <io.h>
#include <limits>
#include <windows.h>
#else
#include <fcntl.h>
#include <unistd.h>
@ -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::size_t>(std::numeric_limits<DWORD>::max())) {
throw std::runtime_error(std::string(name) + ": nbytes exceeds platform limit");
}
if (offset < 0) {
errno = EINVAL;
return -1;
}
const auto handle = reinterpret_cast<HANDLE>(::_get_osfhandle(fd));
if (handle == INVALID_HANDLE_VALUE) {
errno = EBADF;
return -1;
}
OVERLAPPED overlapped{};
overlapped.Offset = static_cast<DWORD>(offset);
overlapped.OffsetHigh = static_cast<DWORD>(static_cast<std::uint64_t>(offset) >> 32u);
DWORD done = 0;
const BOOL ok = write ? ::WriteFile(handle, buf, static_cast<DWORD>(nbyte), &done, &overlapped)
: ::ReadFile(handle, buf, static_cast<DWORD>(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::size_t>(std::numeric_limits<unsigned int>::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<unsigned int>(nbyte));
const int error = errno;
::_close(duped);
errno = error;
return result;
return NativePositioned(fd, buf, nbyte, offset, false, "sceKernelAioSubmitReadCommands");
#else
return static_cast<std::int64_t>(::pread(fd, buf, nbyte, static_cast<off_t>(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::size_t>(std::numeric_limits<unsigned int>::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<unsigned int>(nbyte));
const int error = errno;
::_close(duped);
errno = error;
return result;
return NativePositioned(fd, const_cast<void*>(buf), nbyte, offset, true, "sceKernelAioSubmitWriteCommands");
#else
return static_cast<std::int64_t>(::pwrite(fd, buf, nbyte, static_cast<off_t>(offset)));
#endif

28
core/libs/tests/GuestAio.cpp
View File

@ -3,6 +3,7 @@
#include "prx/libkernel/File/include/FileFlags.hpp"
#include "prx/libkernel/KernelErrors.hpp"
#include <array>
#include <atomic>
#include <chrono>
#include <cstdio>
#include <cstdlib>
@ -11,6 +12,8 @@
#include <fstream>
#include <stdexcept>
#include <string>
#include <thread>
#include <vector>
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<bool> misread{false};
std::vector<std::thread> 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<char, 16> 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);
}