Implement batch AIO completion polling and deletion with per-request results

src/libs/libKernel.cpp

@@ -2828,6 +2828,7 @@ constexpr int32_t KERNEL_AIO_STATE_SUBMITTED  = 1;
 constexpr int32_t KERNEL_AIO_STATE_PROCESSING = 2;
 constexpr int32_t KERNEL_AIO_STATE_COMPLETED  = 3;
 constexpr int32_t KERNEL_AIO_STATE_ABORTED    = 4;
+constexpr int32_t KERNEL_AIO_STATE_NOTIFIED   = 0x10000;
 constexpr int32_t KERNEL_AIO_MAX_QUEUE        = 512;
 constexpr int32_t KERNEL_AIO_MAX_REQUESTS     = 128;
 
@@ -2940,10 +2941,31 @@ int KYTY_SYSV_ABI KernelAioPollRequest(int32_t id, int32_t* state) {
 	}
 
 	if (!kernel_aio_is_valid_id(id)) {
-		return LibKernel::KERNEL_ERROR_EINVAL;
+		*state = LibKernel::KERNEL_ERROR_ESRCH;
+		return OK;
 	}
 
-	*state = g_kernel_aio_state[id].load(std::memory_order_acquire);
+	auto current = g_kernel_aio_state[id].load(std::memory_order_acquire);
+	while (current == KERNEL_AIO_STATE_COMPLETED || current == KERNEL_AIO_STATE_ABORTED) {
+		if (g_kernel_aio_state[id].compare_exchange_weak(current, current | KERNEL_AIO_STATE_NOTIFIED,
+		                                                std::memory_order_acq_rel)) {
+			break;
+		}
+	}
+	*state = current == 0 ? LibKernel::KERNEL_ERROR_ESRCH : current;
+	return OK;
+}
+
+int KYTY_SYSV_ABI KernelAioPollRequests(int32_t* ids, int32_t num, int32_t* states) {
+	if (ids == nullptr || states == nullptr) {
+		return LibKernel::KERNEL_ERROR_EFAULT;
+	}
+	if (num <= 0 || num > KERNEL_AIO_MAX_REQUESTS) {
+		return LibKernel::KERNEL_ERROR_EINVAL;
+	}
+	for (int32_t i = 0; i < num; i++) {
+		KernelAioPollRequest(ids[i], &states[i]);
+	}
 	return OK;
 }
 
@@ -2955,7 +2977,7 @@ int KYTY_SYSV_ABI KernelAioWaitRequest(int32_t id, int32_t* state, uint32_t* use
 	}
 
 	if (!kernel_aio_is_valid_id(id)) {
-		return LibKernel::KERNEL_ERROR_EINVAL;
+		return KernelAioPollRequest(id, state);
 	}
 
 	uint32_t waited  = 0;
@@ -2971,8 +2993,7 @@ int KYTY_SYSV_ABI KernelAioWaitRequest(int32_t id, int32_t* state, uint32_t* use
 		current = g_kernel_aio_state[id].load(std::memory_order_acquire);
 	}
 
-	*state = current;
+	return KernelAioPollRequest(id, state);
-	return OK;
 }
 
 int KYTY_SYSV_ABI KernelAioDeleteRequest(int32_t id, int32_t* ret) {
@@ -2983,12 +3004,35 @@ int KYTY_SYSV_ABI KernelAioDeleteRequest(int32_t id, int32_t* ret) {
 	}
 
 	if (!kernel_aio_is_valid_id(id)) {
-		return LibKernel::KERNEL_ERROR_EINVAL;
+		*ret = LibKernel::KERNEL_ERROR_ESRCH;
+		return OK;
 	}
 
-	g_kernel_aio_state[id].store(KERNEL_AIO_STATE_ABORTED, std::memory_order_release);
+	auto current = g_kernel_aio_state[id].load(std::memory_order_acquire);
-	*ret = OK;
+	for (;;) {
+		const auto state = current & ~KERNEL_AIO_STATE_NOTIFIED;
+		if (state != KERNEL_AIO_STATE_COMPLETED && state != KERNEL_AIO_STATE_ABORTED) {
+			*ret = current == 0 ? LibKernel::KERNEL_ERROR_ESRCH : LibKernel::KERNEL_ERROR_EBUSY;
+			break;
+		}
+		if (g_kernel_aio_state[id].compare_exchange_weak(current, 0, std::memory_order_acq_rel)) {
+			*ret = OK;
+			break;
+		}
+	}
+	return OK;
+}
 
+int KYTY_SYSV_ABI KernelAioDeleteRequests(int32_t* ids, int32_t num, int32_t* rets) {
+	if (ids == nullptr || rets == nullptr) {
+		return LibKernel::KERNEL_ERROR_EFAULT;
+	}
+	if (num <= 0 || num > KERNEL_AIO_MAX_REQUESTS) {
+		return LibKernel::KERNEL_ERROR_EINVAL;
+	}
+	for (int32_t i = 0; i < num; i++) {
+		KernelAioDeleteRequest(ids[i], &rets[i]);
+	}
 	return OK;
 }
 
@@ -3326,8 +3370,10 @@ LIB_DEFINE(InitLibKernel_1) {
 	LIB_FUNC("kbw4UHHSYy0", LibKernel::pthread_cxa_finalize);
 	LIB_FUNC("lLMT9vJAck0", LibKernel::clock_gettime);
 	LIB_FUNC("5TgME6AYty4", KernelAioDeleteRequest);
+	LIB_FUNC("Ft3EtsZzAoY", KernelAioDeleteRequests);
 	LIB_FUNC("HgX7+AORI58", KernelAioSubmitReadCommands);
 	LIB_FUNC("2pOuoWoCxdk", KernelAioPollRequest);
+	LIB_FUNC("o7O4z3jwKzo", KernelAioPollRequests);
 	LIB_FUNC("KOF-oJbQVvc", KernelAioWaitRequest);
 	LIB_FUNC("XQ8C8y+de+E", KernelAioSubmitWriteCommands);
 	LIB_FUNC("nu4a0-arQis", KernelAioInitializeParam);

tests/KernelFileSystemTests.cpp

@@ -35,6 +35,10 @@ namespace Libs::LibKernelApr {
 void InitLibKernel_1_Apr(Loader::SymbolDatabase *symbols);
 }
 
+namespace Libs {
+void InitLibKernel_1(Loader::SymbolDatabase *symbols);
+}
+
 namespace Libs::LibAmpr {
 void InitAmpr_1(Loader::SymbolDatabase *symbols);
 }
@@ -132,6 +136,82 @@ void TestSaveOpenVisibility() {
   Check(FileSystem::KernelClose(truncated) == OK, "close truncated save file");
 }
 
+void TestAioBatches() {
+  namespace Kernel = Libs::LibKernel;
+  Loader::SymbolDatabase symbols;
+  Libs::InitLibKernel_1(&symbols);
+  const auto find = [&](const char *nid) {
+    const auto *symbol = symbols.FindByNid(nid, Loader::SymbolType::Func);
+    Check(symbol != nullptr, "AIO exports resolve");
+    return symbol->vaddr;
+  };
+  struct Result { int64_t return_value; uint32_t state; };
+  struct Request {
+    int64_t offset;
+    size_t size;
+    void *buffer;
+    Result *result;
+    int32_t fd;
+  };
+  using Submit = int (KYTY_SYSV_ABI *)(Request *, int32_t, int32_t, int32_t *);
+  using Batch = int (KYTY_SYSV_ABI *)(int32_t *, int32_t, int32_t *);
+  using Single = int (KYTY_SYSV_ABI *)(int32_t, int32_t *);
+  using Wait = int (KYTY_SYSV_ABI *)(int32_t, int32_t *, uint32_t *);
+  const auto submit = reinterpret_cast<Submit>(find("HgX7+AORI58"));
+  const auto poll = reinterpret_cast<Batch>(find("o7O4z3jwKzo"));
+  const auto erase = reinterpret_cast<Batch>(find("Ft3EtsZzAoY"));
+  const auto poll_one = reinterpret_cast<Single>(find("2pOuoWoCxdk"));
+  const auto erase_one = reinterpret_cast<Single>(find("5TgME6AYty4"));
+  const auto wait = reinterpret_cast<Wait>(find("KOF-oJbQVvc"));
+  constexpr char Payload[] = "AIO payload";
+  const int fd = FileSystem::KernelOpen("/savedata0/aio.dat", 0x602, 0777);
+  Check(fd >= 3 && FileSystem::KernelWrite(fd, Payload, sizeof(Payload)) == sizeof(Payload),
+        "create AIO read fixture");
+  std::array<int32_t, 3> ids {};
+  std::array<char, 4> buffer {};
+  Result result {};
+  Request request {2, buffer.size(), buffer.data(), &result, fd};
+  for (int i = 0; i < 2; ++i) {
+    Check(submit(&request, 1, 2, &ids[i]) == OK &&
+              result.return_value == buffer.size() && result.state == 3 &&
+              std::memcmp(buffer.data(), Payload + 2, buffer.size()) == 0,
+          "AIO submission reads bytes at the requested offset");
+  }
+  ids[2] = -1;
+  std::array<int32_t, 3> states {-1, -1, -1};
+  Check(poll(ids.data(), ids.size(), states.data()) == OK &&
+            states[0] == 3 && states[1] == 3 && states[2] == Kernel::KERNEL_ERROR_ESRCH,
+        "batch poll writes every state and per-request invalid-ID error");
+  Check(poll_one(ids[0], &states[0]) == OK && states[0] == (3 | 0x10000) &&
+            poll(ids.data(), 2, states.data()) == OK &&
+            states[0] == (3 | 0x10000) && states[1] == (3 | 0x10000) &&
+            wait(ids[0], &states[0], nullptr) == OK && states[0] == (3 | 0x10000),
+        "single and batch polls and waits share completion notification state");
+  Check(erase(ids.data(), ids.size(), states.data()) == OK &&
+            states[0] == OK && states[1] == OK && states[2] == Kernel::KERNEL_ERROR_ESRCH,
+        "batch deletion writes per-request results");
+  Check(poll_one(ids[0], &states[0]) == OK && states[0] == Kernel::KERNEL_ERROR_ESRCH &&
+            erase_one(ids[1], &states[1]) == OK && states[1] == Kernel::KERNEL_ERROR_ESRCH,
+        "deleted AIO IDs are invalid for single-request APIs");
+  for (const auto batch : {poll, erase}) {
+    states.fill(42);
+    Check(batch(nullptr, 1, states.data()) == Kernel::KERNEL_ERROR_EFAULT &&
+              batch(ids.data(), 1, nullptr) == Kernel::KERNEL_ERROR_EFAULT &&
+              batch(ids.data(), 0, states.data()) == Kernel::KERNEL_ERROR_EINVAL &&
+              batch(ids.data(), 129, states.data()) == Kernel::KERNEL_ERROR_EINVAL &&
+              states == std::array<int32_t, 3> {42, 42, 42},
+          "invalid batch arguments do not modify outputs");
+    std::array<int32_t, 128> invalid_ids {};
+    std::array<int32_t, 128> errors {};
+    Check(batch(invalid_ids.data(), invalid_ids.size(), errors.data()) == OK &&
+              std::all_of(errors.begin(), errors.end(), [](int32_t error) {
+                return error == Kernel::KERNEL_ERROR_ESRCH;
+              }),
+          "maximum-sized batch returns all invalid-ID errors");
+  }
+  Check(FileSystem::KernelClose(fd) == OK, "close AIO read fixture");
+}
+
 void CheckMountRoot(const std::filesystem::path &root) {
   Common::File cache;
   Check(cache.Create(root / "rpf.cache"), "create directory listing fixture");
@@ -836,6 +916,7 @@ int main(int, char**) {
   CheckAprPaths(temporary.Path());
   FileSystem::Mount(temporary.Path(), "/savedata0");
   TestSaveOpenVisibility();
+  TestAioBatches();
   CheckSaveRename(temporary.Path(), "first-save");
   CheckSaveRename(temporary.Path(), "replacement-save");
   FileSystem::Shutdown();