feat(libkernel): support EV_CLEAR on a kqueue watched by another kqueue

core/libs/prx/libkernel/CMakeLists.txt

@@ -1514,9 +1514,11 @@ if(BUILD_TESTING)
     configure_windows_unwind(guest_kqueue_tests)
     add_test(NAME guest_kqueue COMMAND guest_kqueue_tests)
     set_tests_properties(guest_kqueue PROPERTIES TIMEOUT 20)
+    add_test(NAME guest_kqueue_cleared_race COMMAND guest_kqueue_tests cleared-race)
+    set_tests_properties(guest_kqueue_cleared_race PROPERTIES TIMEOUT 60)
     if(WIN32)
         target_link_libraries(guest_kqueue_tests PRIVATE advapi32)
-        set_tests_properties(guest_kqueue PROPERTIES ENVIRONMENT_MODIFICATION
+        set_tests_properties(guest_kqueue guest_kqueue_cleared_race PROPERTIES ENVIRONMENT_MODIFICATION
             "PATH=path_list_prepend:$<TARGET_FILE_DIR:libkernel>")
     endif()
     add_test_executable(guest_socket_descriptors_tests ${LIBS_INCLUDE_DIR}/tests/GuestSocketDescriptors.cpp)

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

@@ -66,6 +66,7 @@ void KernelEqueuePrivate::TriggerExpiredTimers(uint64_t nowNs) {
                 ev.event.data += static_cast<intptr_t>(count);
                 ev.deadlineNs += count * ev.intervalNs;
             }
+            if (!ev.triggered) ++m_activations;
             ev.triggered = true;
         }
     }
@@ -91,7 +92,9 @@ bool KernelEqueuePrivate::PollEvents() {
     bool polled = false;
     for (auto& ev : m_events) {
         if (ev.enabled && ev.filter.pollFunc != nullptr) {
+            const bool was = ev.triggered;
             ev.triggered = ev.filter.pollFunc(&ev);
+            if (!was && ev.triggered) ++m_activations;
             polled = true;
         }
     }
@@ -220,6 +223,7 @@ int KernelEqueuePrivate::CollectTriggered(KernelEvent* ev, int num) {
             e.event = e.pendingEvents.front();
             e.pendingEvents.pop_front();
             e.triggered = true;
+            ++m_activations;
         }
         it = erase ? m_events.erase(it) : std::next(it);
     }
@@ -290,6 +294,17 @@ std::size_t KernelEqueuePrivate::CountTriggeredEvents() {
     return static_cast<std::size_t>(std::count_if(m_events.begin(), m_events.end(), [](const KernelEqueueEvent& event) { return event.enabled && event.triggered; }));
 }
 
+KernelEqueuePrivate::TriggeredSnapshot KernelEqueuePrivate::SnapshotTriggered() {
+    std::unique_lock lock(m_mutex);
+    if (m_closed) {
+        return {0, m_activations};
+    }
+    TriggerExpiredTimers(MonotonicNs());
+    PollEvents();
+    const auto count = static_cast<std::size_t>(std::count_if(m_events.begin(), m_events.end(), [](const KernelEqueueEvent& event) { return event.enabled && event.triggered; }));
+    return {count, m_activations};
+}
+
 std::vector<std::shared_ptr<KernelEqueuePrivate>> KernelEqueuePrivate::WatchedQueues() {
     std::unique_lock lock(m_mutex);
     std::vector<std::shared_ptr<KernelEqueuePrivate>> queues;
@@ -371,6 +386,7 @@ int KernelEqueuePrivate::AddEvent(const KernelEqueueEvent& event) {
         }
     } else {
         m_events.push_back(event);
+        if (event.triggered) ++m_activations;
     }
     m_cond.NotifyOne();
     WakePollers();
@@ -391,11 +407,13 @@ int KernelEqueuePrivate::TriggerEvent(uintptr_t ident, int16_t filter, void* tri
         return SCE_KERNEL_ERROR_ENOENT;
     }
     if (it->enabled) {
+        const bool was = it->triggered;
         if (it->filter.triggerFunc != nullptr) {
             it->filter.triggerFunc(&*it, triggerData);
         } else {
             it->triggered = true;
         }
+        if (!was && it->triggered) ++m_activations;
     }
     m_cond.NotifyOne();
     WakePollers();
@@ -422,6 +440,14 @@ int KernelEqueuePrivate::EditEvent(uintptr_t ident, int16_t filter, const std::f
     if (it == m_events.end()) {
         return SCE_KERNEL_ERROR_ENOENT;
     }
+    struct Activation {
+        KernelEqueuePrivate& queue;
+        KernelEqueueEvent& event;
+        bool active;
+        ~Activation() {
+            if (!active && event.enabled && event.triggered) ++queue.m_activations;
+        }
+    } activation{*this, *it, it->enabled && it->triggered};
     edit(*it);
     m_cond.NotifyOne();
     WakePollers();

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

@@ -90,6 +90,11 @@ class KernelEqueuePrivate {
     int GetTriggeredEvents(KernelEvent* ev, int num);
     bool HasTriggeredEvents();
     std::size_t CountTriggeredEvents();
+    struct TriggeredSnapshot {
+        std::size_t count = 0;
+        std::uint64_t activations = 0;
+    };
+    TriggeredSnapshot SnapshotTriggered();
     std::vector<std::shared_ptr<KernelEqueuePrivate>> WatchedQueues();
     int WaitForEvents(KernelEvent* ev, int num, uint32_t micros);
     int WaitForEventsUntil(KernelEvent* ev, int num, std::uint64_t deadlineNanos);
@@ -118,6 +123,7 @@ class KernelEqueuePrivate {
     KernelEqueue m_handle;
     bool m_reportOncePerWait;
     bool m_closed = false;
+    std::uint64_t m_activations = 0;
 };
 
 using KernelEqueueRef = std::shared_ptr<KernelEqueuePrivate>;

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

@@ -196,6 +196,33 @@ bool PollKqueue(KernelEqueueEvent* e) {
     return pending != 0;
 }
 
+struct ClearedKqueueNote {
+    std::uint64_t seen = 0;
+    bool active = false;
+};
+
+bool PollClearedKqueue(KernelEqueueEvent* e) {
+    auto& note = *static_cast<ClearedKqueueNote*>(e->filter.data);
+    const auto snapshot = e->filter.watched->SnapshotTriggered();
+    if (snapshot.activations != note.seen) {
+        note.seen = snapshot.activations;
+        note.active = true;
+    }
+    if (snapshot.count == 0) {
+        note.active = false;
+    } else if (note.active) {
+        e->event.data = static_cast<std::intptr_t>(snapshot.count);
+    }
+    return note.active;
+}
+
+void ResetClearedKqueue(KernelEqueueEvent* e) {
+    static_cast<ClearedKqueueNote*>(e->filter.data)->active = false;
+    e->triggered = false;
+    e->event.fflags = 0;
+    e->event.data = 0;
+}
+
 bool KqueueReaches(const KernelEqueueRef& from, const KernelEqueuePrivate* target, std::vector<const KernelEqueuePrivate*>& visited) {
     if (from.get() == target) {
         return true;
@@ -213,9 +240,6 @@ bool KqueueReaches(const KernelEqueueRef& from, const KernelEqueuePrivate* targe
 }
 
 int AddKqueueKnote(const KernelEqueueRef& queue, const KernelEqueueRef& watched, KernelEvent& change, std::uint16_t flags) {
-    if ((flags & EV_CLEAR) != 0) {
-        NotImplemented_nid_no_patch("kevent: EV_CLEAR on a kqueue descriptor");
-    }
     std::lock_guard lock(g_kqueueWatchMutex);
     std::vector<const KernelEqueuePrivate*> visited;
     if (KqueueReaches(watched, queue.get(), visited)) {
@@ -228,6 +252,17 @@ int AddKqueueKnote(const KernelEqueueRef& queue, const KernelEqueueRef& watched,
     event.event.data = 0;
     event.filter.pollFunc = PollKqueue;
     event.filter.watched = watched;
+    if ((flags & EV_CLEAR) != 0) {
+        auto note = std::make_shared<ClearedKqueueNote>();
+        const auto snapshot = watched->SnapshotTriggered();
+        note->seen = snapshot.activations;
+        note->active = snapshot.count != 0;
+        event.event.data = static_cast<std::intptr_t>(snapshot.count);
+        event.filter.data = note.get();
+        event.filter.owner = std::move(note);
+        event.filter.pollFunc = PollClearedKqueue;
+        event.filter.resetFunc = ResetClearedKqueue;
+    }
     change.fflags = 0;
     change.data = 0;
     return GuestErrno(queue->AddEvent(event));
@@ -354,6 +389,9 @@ int ModifyKnote(const KernelEqueueRef& queue, KernelEvent& change) {
         if (event.enabled && event.filter.pollFunc == PollEdge) {
             static_cast<EdgeWatch::Watch*>(event.filter.data)->armed = true;
         }
+        if (event.enabled && event.filter.pollFunc == PollClearedKqueue) {
+            static_cast<ClearedKqueueNote*>(event.filter.data)->active = true;
+        }
         if (change.filter == EVFILT_USER) {
             TouchUserKnote(event, change);
         } else if (change.filter == EVFILT_VNODE) {

core/libs/tests/GuestKqueue.cpp

@@ -14,6 +14,8 @@
 #ifdef _WIN32
 #include <aclapi.h>
 #include <windows.h>
+#elif defined(__linux__)
+#include <sched.h>
 #endif
 
 extern "C" {
@@ -484,8 +486,8 @@ static void WatchKqueue() {
 
     RequireFailure(Apply(outer, Change(watched, EvfiltWrite, EvAdd)), GuestEinval);
     RequireFailure(Apply(outer, Change(watched, EvfiltVnode, EvAdd, NoteWrite)), GuestEinval);
-    Require(ThrownMessage([outer, watched] { Apply(outer, Change(watched, EvfiltRead, EvAdd | EvClear)); }).find("EV_CLEAR on a kqueue descriptor") != std::string::npos);
-    RequireFailure(Apply(outer, Change(watched, EvfiltRead, EvDelete)), GuestEnoent);
+    Require(Apply(outer, Change(watched, EvfiltRead, EvAdd | EvClear)) == 0);
+    Require(Apply(outer, Change(watched, EvfiltRead, EvDelete)) == 0);
     Require(Apply(outer, Change(watched, EvfiltRead, EvAdd)) == 0);
     Require(ThrownMessage([inner, outer] { Apply(inner, Change(static_cast<std::uintptr_t>(outer), EvfiltRead, EvAdd)); }).find("watches the kqueue") != std::string::npos);
     Require(Apply(inner, Change(5, EvfiltUser, 0, NoteTrigger)) == 0);
@@ -497,6 +499,106 @@ static void WatchKqueue() {
     Require(close_nid_postfix(outer) == 0);
 }
 
+static void ClearedKqueue() {
+    const int inner = kqueue_nid_postfix();
+    const int outer = kqueue_nid_postfix();
+    Require(inner >= 0 && outer >= 0);
+    const auto watched = static_cast<std::uintptr_t>(inner);
+    KernelEvent events[4]{};
+    Require(Apply(inner, Change(7, EvfiltUser, EvAdd)) == 0);
+    Require(Apply(inner, Change(8, EvfiltUser, EvAdd)) == 0);
+    Require(Apply(outer, Change(watched, EvfiltRead, EvAdd | EvClear)) == 0);
+    Require(Poll(outer, events, 4) == 0);
+
+    Require(Apply(inner, Change(7, EvfiltUser, 0, NoteTrigger)) == 0);
+    Require(Poll(outer, events, 4) == 1 && events[0].ident == watched && events[0].data == 1 && (events[0].flags & EvClear) != 0);
+    Require(Poll(outer, events, 4) == 0);
+    Require(Poll(inner, events, 4) == 1);
+    Require(Poll(outer, events, 4) == 0);
+
+    Require(Apply(inner, Change(8, EvfiltUser, 0, NoteTrigger)) == 0);
+    Require(Poll(outer, events, 4) == 1 && events[0].data == 2);
+    Require(Poll(outer, events, 4) == 0);
+
+    Require(Apply(outer, Change(watched, EvfiltRead, EvDelete)) == 0);
+    Require(Apply(outer, Change(watched, EvfiltRead, EvAdd | EvClear)) == 0);
+    Require(Poll(outer, events, 4) == 1 && events[0].data == 2);
+    Require(Poll(outer, events, 4) == 0);
+    Require(close_nid_postfix(outer) == 0 && close_nid_postfix(inner) == 0);
+}
+
+static void ClearedKqueueRechecksTheWatchedCount() {
+    const int inner = kqueue_nid_postfix();
+    const int outer = kqueue_nid_postfix();
+    Require(inner >= 0 && outer >= 0);
+    const auto watched = static_cast<std::uintptr_t>(inner);
+    KernelEvent events[4]{};
+    Require(Apply(inner, Change(7, EvfiltUser, EvAdd | EvClear)) == 0);
+    Require(Apply(inner, Change(8, EvfiltUser, EvAdd | EvClear)) == 0);
+    Require(Apply(outer, Change(1, EvfiltUser, EvAdd | EvClear)) == 0);
+    Require(Apply(outer, Change(watched, EvfiltRead, EvAdd | EvClear)) == 0);
+
+    Require(Apply(inner, Change(7, EvfiltUser, 0, NoteTrigger)) == 0);
+    Require(Apply(outer, Change(1, EvfiltUser, 0, NoteTrigger)) == 0);
+    Require(Poll(outer, events, 1) == 1 && events[0].ident == 1 && events[0].filter == EvfiltUser);
+    Require(Poll(inner, events, 4) == 1 && events[0].ident == 7);
+    Require(Poll(outer, events, 4) == 0);
+
+    Require(Apply(inner, Change(7, EvfiltUser, 0, NoteTrigger)) == 0);
+    Require(Apply(inner, Change(8, EvfiltUser, 0, NoteTrigger)) == 0);
+    Require(Apply(outer, Change(1, EvfiltUser, 0, NoteTrigger)) == 0);
+    Require(Poll(outer, events, 1) == 1 && events[0].ident == 1);
+    Require(Poll(inner, events, 1) == 1);
+    Require(Poll(outer, events, 4) == 1 && events[0].ident == watched && events[0].data == 1);
+    Require(Poll(outer, events, 4) == 0);
+
+    Require(Apply(outer, Change(watched, EvfiltRead, EvAdd | EvClear)) == 0);
+    Require(Poll(outer, events, 4) == 1 && events[0].ident == watched && events[0].data == 1);
+    Require(Poll(outer, events, 4) == 0);
+    Require(Apply(outer, Change(watched, EvfiltRead, EvDisable)) == 0);
+    Require(Apply(outer, Change(watched, EvfiltRead, EvEnable)) == 0);
+    Require(Poll(outer, events, 4) == 1 && events[0].ident == watched && events[0].data == 1);
+    Require(Poll(inner, events, 4) == 1);
+    Require(Apply(outer, Change(watched, EvfiltRead, EvAdd | EvClear)) == 0);
+    Require(Poll(outer, events, 4) == 0);
+    Require(close_nid_postfix(outer) == 0 && close_nid_postfix(inner) == 0);
+}
+
+static void RunOnOneCpuOnLinux() {
+#ifdef __linux__
+    cpu_set_t allowed;
+    CPU_ZERO(&allowed);
+    Require(sched_getaffinity(0, sizeof(allowed), &allowed) == 0);
+    int first = 0;
+    while (first < CPU_SETSIZE && !CPU_ISSET(first, &allowed)) ++first;
+    Require(first < CPU_SETSIZE);
+    cpu_set_t one;
+    CPU_ZERO(&one);
+    CPU_SET(first, &one);
+    Require(sched_setaffinity(0, sizeof(one), &one) == 0);
+#endif
+}
+
+static void ClearedKqueueKeepsConcurrentActivations() {
+    RunOnOneCpuOnLinux();
+    const int inner = kqueue_nid_postfix();
+    const int outer = kqueue_nid_postfix();
+    Require(inner >= 0 && outer >= 0);
+    const auto watched = static_cast<std::uintptr_t>(inner);
+    KernelEvent events[4]{};
+    Require(Apply(inner, Change(7, EvfiltUser, EvAdd | EvClear)) == 0);
+    for (std::uintptr_t filler = 100; filler < 20100; ++filler) Require(Apply(inner, Change(filler, EvfiltUser, EvAdd)) == 0);
+    Require(Apply(outer, Change(watched, EvfiltRead, EvAdd | EvClear)) == 0);
+    for (int round = 0; round < 5000; ++round) {
+        std::thread trigger([inner] { Require(Apply(inner, Change(7, EvfiltUser, 0, NoteTrigger)) == 0); });
+        const int reported = Wait(outer, events, 4);
+        trigger.join();
+        Require(reported == 1 && events[0].ident == watched);
+        Require(Poll(inner, events, 4) == 1);
+    }
+    Require(close_nid_postfix(outer) == 0 && close_nid_postfix(inner) == 0);
+}
+
 static void LowWaterMarks() {
     const int kq = kqueue_nid_postfix();
     int pair[2];
@@ -1133,7 +1235,11 @@ static void StandardInput() {
     Require(close_nid_postfix(kq) == 0);
 }
 
-int main() {
+int main(int argc, char** argv) {
+    if (argc == 2 && std::string(argv[1]) == "cleared-race") {
+        ClearedKqueueKeepsConcurrentActivations();
+        return 0;
+    }
     StandardInput();
     CreateAndClose();
     Timers();
@@ -1154,6 +1260,8 @@ int main() {
     DisableAndDispatch();
     Errors();
     LowWaterMarks();
+    ClearedKqueue();
+    ClearedKqueueRechecksTheWatchedCount();
     FileWatch();
 #ifndef _WIN32
     RenameAndDeleteWatchedFile();