monitoring arbitrary kevents

trio/_core/_io_kqueue.py

@@ -79,19 +79,32 @@ def current_kqueue(self):
 
     @_public
     @contextmanager
-    def monitor_kevent(self, ident, filter):
+    def monitor_kevent(self, ident, filter, fflags=0):
         key = (ident, filter)
         if key in self._registered:
             raise _core.ResourceBusyError(
                 "attempt to register multiple listeners for same "
                 "ident/filter pair"
             )
+
+        flags = select.KQ_EV_ADD
+        event = select.kevent(ident, filter, flags, fflags)
+        self._kqueue.control([event], 0)
+
         q = _core.UnboundedQueue()
         self._registered[key] = q
         try:
             yield q
         finally:
-            del self._registered[key]
+            try:
+                del self._registered[key]
+            except KeyError:
+                # was a one shot event and already got deleted?
+                pass
+            else:
+                flags = select.KQ_EV_DELETE
+                event = select.kevent(ident, filter, flags)
+                self._kqueue.control([event], 0)
 
     @_public
     async def wait_kevent(self, ident, filter, abort_func):

trio/_core/tests/test_kqueue.py

@@ -0,0 +1,25 @@
+import select
+import subprocess
+
+import pytest
+
+from ... import _core
+
+use_kqueue = hasattr(select, 'kqueue')
+pytestmark = pytest.mark.skipif(not use_kqueue, reason="kqueue platforms only")
+
+
+async def test_kqueue_monitor_proc_event():
+    p = subprocess.Popen(['sleep', '0.1'])
+    ident = p.pid
+    filter = select.KQ_FILTER_PROC
+    fflags = select.KQ_NOTE_EXIT
+    with _core.monitor_kevent(ident, filter, fflags) as q:
+        async for batch in q:
+            for evt in batch:
+                assert evt.ident == ident
+                assert evt.filter == filter
+                assert evt.fflags == fflags
+                assert evt.data == 0  # exit code
+                # we only expect one event for KQ_NOTE_EXIT
+                return