feat(libkernel): support kevent read and write events on regular files

core/libs/prx/libkernel/Equeue/Equeue.hpp

@@ -120,6 +120,9 @@ KernelEqueueEvent EqueueDescriptorEvent(int descriptor, int16_t filter, uint16_t
 void RequireVnodeNotes(uint32_t fflags);
 void SetVnodeNotes(KernelEqueueEvent& event, uint32_t fflags);
 int AddVnodeEvent(const KernelEqueueRef& queue, int descriptor, uint32_t fflags, uint16_t flags, void* udata);
+bool IsFileDataEvent(const KernelEqueueEvent& event);
+void SetFileDataNotes(KernelEqueueEvent& event, uint32_t fflags);
+int AddFileDataEvent(const KernelEqueueRef& queue, int descriptor, int16_t filter, uint32_t fflags, uint16_t flags, void* udata);
 extern "C" {
 
 KernelEqueueRef EqueuePin_nid_postfix(KernelEqueue eq);

core/libs/prx/libkernel/Equeue/src/FileEvent.cpp

@@ -1,5 +1,6 @@
 #include "prx/libkernel/Equeue/Equeue.hpp"
 #include "prx/libc/include/General.hpp"
+#include "prx/libkernel/File/include/ArchivedFile.hpp"
 #include "prx/libkernel/File/include/File.hpp"
 #include "prx/libkernel/Socket/include/SocketRuntime.hpp"
 
@@ -52,7 +53,7 @@ constexpr std::uint32_t MAPPED_NOTES = NOTE_DELETE | NOTE_WRITE | NOTE_EXTEND |
 
 class VnodeWatch {
 public:
-    VnodeWatch(const KernelEqueueRef& owner, std::uintptr_t descriptor, std::uint32_t notes) : queue(owner), ident(descriptor), interest(notes) {
+    VnodeWatch(const KernelEqueueRef& owner, std::uintptr_t descriptor, std::int16_t filter, std::uint32_t notes) : queue(owner), ident(descriptor), filter(filter), interest(notes) {
 #ifdef _WIN32
         stopEvent = ::CreateEventW(nullptr, TRUE, FALSE, nullptr);
         if (stopEvent == nullptr) {
@@ -118,6 +119,7 @@ class VnodeWatch {
 private:
     std::weak_ptr<KernelEqueuePrivate> queue;
     std::uintptr_t ident;
+    std::int16_t filter;
     std::atomic<std::uint32_t> interest;
     std::atomic<bool> stopped{false};
 #ifdef _WIN32
@@ -144,7 +146,7 @@ void VnodeWatch::Deliver(std::uint32_t hint) {
         return;
     }
     VnodeHint message{this, hint};
-    owner->TriggerEvent(ident, EVFILT_VNODE, &message);
+    owner->TriggerEvent(ident, filter, &message);
 }
 
 #ifdef _WIN32
@@ -682,6 +684,45 @@ void StopVnode(KernelEqueue, KernelEqueueEvent* event) {
     static_cast<VnodeWatch*>(event->filter.data)->Stop();
 }
 
+constexpr std::uint32_t NOTE_FILE_POLL = 0x0002;
+constexpr int GUEST_SEEK_CUR = 1;
+
+struct FileDataNote {
+    std::atomic<std::uint32_t> notes;
+    std::shared_ptr<VnodeWatch> watch;
+};
+
+bool PollFileRead(KernelEqueueEvent* event) {
+    const int descriptor = static_cast<int>(event->event.ident);
+    FileStat status{};
+    if (sceKernelFstat(descriptor, &status) != 0) {
+        return false;
+    }
+    const std::int64_t offset = sceKernelLseek(descriptor, 0, GUEST_SEEK_CUR);
+    if (offset < 0) {
+        return false;
+    }
+    event->event.data = static_cast<std::intptr_t>(status.st_size - offset);
+    const auto* note = static_cast<const FileDataNote*>(event->filter.data);
+    return (note->notes.load() & NOTE_FILE_POLL) != 0 || event->event.data != 0;
+}
+
+bool PollFileWrite(KernelEqueueEvent* event) {
+    event->event.data = 0;
+    return true;
+}
+
+void TriggerFileRead(KernelEqueueEvent* event, void*) {
+    event->triggered = PollFileRead(event);
+}
+
+void StopFileData(KernelEqueue, KernelEqueueEvent* event) {
+    const auto* note = static_cast<const FileDataNote*>(event->filter.data);
+    if (note->watch) {
+        note->watch->Stop();
+    }
+}
+
 int VnodeDescriptorError(int descriptor) {
     if (descriptor < 0) {
         return GUEST_EBADF;
@@ -718,7 +759,7 @@ int AddVnodeEvent(const KernelEqueueRef& queue, int descriptor, std::uint32_t ff
     }
     std::shared_ptr<VnodeWatch> watch;
     try {
-        watch = std::make_shared<VnodeWatch>(queue, static_cast<std::uintptr_t>(descriptor), fflags);
+        watch = std::make_shared<VnodeWatch>(queue, static_cast<std::uintptr_t>(descriptor), EVFILT_VNODE, fflags);
     } catch (...) {
         CloseHostWatch(host);
         throw;
@@ -747,6 +788,59 @@ int AddVnodeEvent(const KernelEqueueRef& queue, int descriptor, std::uint32_t ff
     return 0;
 }
 
+bool IsFileDataEvent(const KernelEqueueEvent& event) {
+    return event.filter.deleteEventFunc == StopFileData;
+}
+
+void SetFileDataNotes(KernelEqueueEvent& event, std::uint32_t fflags) {
+    static_cast<FileDataNote*>(event.filter.data)->notes.store(fflags);
+}
+
+int AddFileDataEvent(const KernelEqueueRef& queue, int descriptor, std::int16_t filter, std::uint32_t fflags, std::uint16_t flags, void* udata) {
+    auto note = std::make_shared<FileDataNote>();
+    note->notes.store(fflags);
+    KernelEqueueEvent event{};
+    event.event.ident = static_cast<std::uintptr_t>(descriptor);
+    event.event.filter = filter;
+    event.event.flags = flags;
+    event.event.udata = udata;
+    event.filter.data = note.get();
+    event.filter.owner = note;
+    event.filter.deleteEventFunc = StopFileData;
+    if (filter == EVFILT_WRITE) {
+        event.filter.pollFunc = PollFileWrite;
+        return queue->AddEvent(event) & 0xffff;
+    }
+    event.filter.pollFunc = PollFileRead;
+    event.filter.triggerFunc = TriggerFileRead;
+    if (File::IsArchivedFile(descriptor)) {
+        return queue->AddEvent(event) & 0xffff;
+    }
+    HostWatch host;
+    if (const int error = OpenHostWatch(descriptor, MAPPED_NOTES, host)) {
+        return error;
+    }
+    try {
+        note->watch = std::make_shared<VnodeWatch>(queue, static_cast<std::uintptr_t>(descriptor), EVFILT_READ, MAPPED_NOTES);
+    } catch (...) {
+        CloseHostWatch(host);
+        throw;
+    }
+    const int result = queue->AddEvent(event);
+    bool registered = false;
+    if (result == EQUEUE_OK) {
+        queue->EditEvent(event.event.ident, EVFILT_READ, [&registered, &note](KernelEqueueEvent& added) {
+            registered = added.filter.data == note.get();
+        });
+    }
+    if (!registered) {
+        CloseHostWatch(host);
+        return result == EQUEUE_OK ? 0 : result & 0xffff;
+    }
+    std::thread(RunHostWatch, note->watch, std::move(host)).detach();
+    return 0;
+}
+
 extern "C" {
 
 int APS5_VABI sceKernelAddFileEvent(KernelEqueue eq, int fd, int watch, void* udata) {

core/libs/prx/libkernel/Equeue/src/Kqueue.cpp

@@ -152,6 +152,11 @@ int AddStandardInputKnote(const KernelEqueueRef& queue, KernelEvent& change, std
     return GuestErrno(queue->AddEvent(event));
 }
 
+bool IsRegularFile(int descriptor) {
+    FileStat status{};
+    return !GuestSockets::IsOpen(descriptor) && sceKernelFstat(descriptor, &status) == 0 && (status.st_mode & 0170000) == 0100000;
+}
+
 int AddDescriptorKnote(const KernelEqueueRef& queue, KernelEvent& change, std::uint16_t flags) {
     const int descriptor = static_cast<int>(change.ident);
     if (IsKqueueDescriptor(descriptor)) {
@@ -163,6 +168,15 @@ int AddDescriptorKnote(const KernelEqueueRef& queue, KernelEvent& change, std::u
     if (descriptor == 0 && change.filter == EVFILT_READ && File::IsHostStream(descriptor)) {
         return AddStandardInputKnote(queue, change, flags);
     }
+    if (IsRegularFile(descriptor)) {
+        if ((flags & EV_CLEAR) != 0) {
+            NotImplemented_nid_no_patch("kevent: EV_CLEAR on a regular file");
+        }
+        const std::uint32_t notes = change.fflags;
+        change.fflags = 0;
+        change.data = 0;
+        return AddFileDataEvent(queue, descriptor, change.filter, notes, flags, change.udata);
+    }
     const bool socket = GuestSockets::IsOpen(descriptor);
     std::int64_t data = 0;
     bool eof = false;
@@ -270,6 +284,8 @@ int ModifyKnote(const KernelEqueueRef& queue, KernelEvent& change) {
             TouchUserKnote(event, change);
         } else if (change.filter == EVFILT_VNODE) {
             SetVnodeNotes(event, change.fflags);
+        } else if (IsFileDataEvent(event)) {
+            SetFileDataNotes(event, change.fflags);
         }
     }));
 }

core/libs/prx/libkernel/File/include/File.hpp

@@ -18,6 +18,7 @@ std::int64_t APS5_VABI sceKernelRead(int d, void* buf, std::size_t nbytes);
 std::int64_t APS5_VABI sceKernelWrite(int d, const void* buf, std::size_t nbytes);
 std::int64_t APS5_VABI sceKernelLseek(int d, std::int64_t offset, int whence);
 int APS5_VABI sceKernelStat(const char* path, FileStat* sb);
+int APS5_VABI sceKernelFstat(int d, FileStat* sb);
 int APS5_VABI sceKernelUnlink(const char* path);
 int APS5_VABI sceKernelFcntl(int d, int command, ...);
 #ifdef _WIN32

core/libs/prx/libkernel/Socket/src/Udp.cpp

@@ -371,9 +371,11 @@ bool GuestSockets::WaitAny(const std::vector<Interest>& interests, std::uint64_t
             const auto found = sockets.find(interest.descriptor);
             if (found == sockets.end()) {
 #ifdef _WIN32
-                pipes = true;
+                const auto handle = reinterpret_cast<HANDLE>(::_get_osfhandle(interest.descriptor));
+                if (handle != INVALID_HANDLE_VALUE && GetFileType(handle) == FILE_TYPE_PIPE) pipes = true;
 #else
-                entries.push_back({interest.descriptor, ReadMask, 0});
+                struct stat status{};
+                if (::fstat(interest.descriptor, &status) == 0 && S_ISFIFO(status.st_mode)) entries.push_back({interest.descriptor, interest.write ? WriteMask : ReadMask, 0});
 #endif
                 continue;
             }

core/libs/tests/GuestKqueue.cpp

@@ -29,6 +29,7 @@ int APS5_VABI open_nid_postfix(const char*, int, int);
 int APS5_VABI mkdir_nid_postfix(const char*, unsigned short);
 int APS5_VABI rmdir_nid_postfix(const char*);
 std::int64_t APS5_VABI write_nid_postfix(int, const char*, std::int64_t);
+std::int64_t APS5_VABI lseek_nid_postfix(int, std::int64_t, int);
 int APS5_VABI ftruncate_nid_postfix(int, std::int64_t);
 int APS5_VABI unlink_nid_postfix(const char*);
 int APS5_VABI sceKernelRename(const char*, const char*);
@@ -69,6 +70,7 @@ static constexpr std::uint32_t NoteFfcopy = 0xc0000000;
 static constexpr std::uint32_t NoteFfor = 0x80000000;
 static constexpr std::uint32_t NoteUseconds = 0x0004;
 static constexpr std::uint32_t NoteNseconds = 0x0008;
+static constexpr std::uint32_t NoteFilePoll = 0x0002;
 static constexpr std::uint32_t NoteDelete = 0x0001;
 static constexpr std::uint32_t NoteWrite = 0x0002;
 static constexpr std::uint32_t NoteExtend = 0x0004;
@@ -429,6 +431,56 @@ static std::string TemporaryName(const char* stem) {
     return std::string("anyps5-kqueue-") + stem + "-" + std::to_string(std::chrono::steady_clock::now().time_since_epoch().count());
 }
 
+static void RegularFile() {
+    const auto path = TemporaryName("data");
+    const int file = open_nid_postfix(path.c_str(), ReadWrite | Create, 0644);
+    Require(file >= 0);
+    Require(write_nid_postfix(file, "abcd", 4) == 4);
+    Require(lseek_nid_postfix(file, 1, 0) == 1);
+    const int kq = kqueue_nid_postfix();
+    const auto ident = static_cast<std::uintptr_t>(file);
+    int marker = 0;
+    KernelEvent events[4]{};
+    Require(Apply(kq, Change(ident, EvfiltRead, EvAdd, 0, 0, &marker)) == 0);
+    Require(Poll(kq, events, 4) == 1);
+    Require(events[0].ident == ident && events[0].filter == EvfiltRead && events[0].flags == 0 && events[0].fflags == 0);
+    Require(events[0].data == 3 && events[0].udata == &marker);
+    char buffer[4]{};
+    Require(read_nid_postfix(file, buffer, 3) == 3);
+    Require(Poll(kq, events, 4) == 0);
+    Require(lseek_nid_postfix(file, 6, 0) == 6);
+    Require(Poll(kq, events, 4) == 1 && events[0].data == -2);
+    Require(lseek_nid_postfix(file, 4, 0) == 4);
+    Require(Poll(kq, events, 4) == 0);
+    Require(Apply(kq, Change(ident, EvfiltRead, EvAdd, NoteFilePoll)) == 0);
+    Require(Poll(kq, events, 4) == 1 && events[0].data == 0 && events[0].udata == nullptr);
+    Require(Apply(kq, Change(ident, EvfiltRead, EvAdd, NoteLowat, 100, &marker)) == 0);
+    Require(Poll(kq, events, 4) == 0);
+    const int appender = open_nid_postfix(path.c_str(), ReadWrite, 0);
+    Require(appender >= 0 && lseek_nid_postfix(appender, 0, 2) == 4);
+    const auto start = std::chrono::steady_clock::now();
+    std::thread writer([appender] {
+        std::this_thread::sleep_for(std::chrono::milliseconds(50));
+        Require(write_nid_postfix(appender, "ef", 2) == 2);
+    });
+    const KernelTimespec limit{5, 0};
+    Require(kevent_nid_postfix(kq, nullptr, 0, events, 4, &limit) == 1);
+    const auto waited = std::chrono::steady_clock::now() - start;
+    writer.join();
+    Require(waited >= std::chrono::milliseconds(40) && waited < std::chrono::seconds(4));
+    Require(events[0].ident == ident && events[0].data == 2 && events[0].udata == &marker);
+    Require(ftruncate_nid_postfix(appender, 4) == 0);
+    Require(Poll(kq, events, 4) == 0);
+    Require(Throws([kq, ident] { Apply(kq, Change(ident, EvfiltWrite, EvAdd | EvClear)); }));
+    Require(Apply(kq, Change(ident, EvfiltWrite, EvAdd)) == 0);
+    Require(Poll(kq, events, 4) == 1 && events[0].filter == EvfiltWrite && events[0].data == 0);
+    Require(close_nid_postfix(appender) == 0);
+    Require(close_nid_postfix(file) == 0);
+    Require(Poll(kq, events, 4) == 0);
+    Require(unlink_nid_postfix(path.c_str()) == 0);
+    Require(close_nid_postfix(kq) == 0);
+}
+
 static void FileWatch() {
     const auto path = TemporaryName("file");
     const int file = open_nid_postfix(path.c_str(), ReadWrite | Create, 0644);
@@ -742,6 +794,7 @@ int main() {
     Errors();
     LowWaterMarks();
     FileWatch();
+    RegularFile();
 #ifndef _WIN32
     RenameAndDeleteWatchedFile();
 #endif