Method _wait_kqueue

Lib/subprocess.py:2131–2157

Wait for PID to terminate using kqueue(). macOS and BSD only.

(self, timeout)

                os.close(pidfd)

        def _wait_kqueue(self, timeout):
            """Wait for PID to terminate using kqueue(). macOS and BSD only."""
            if not _CAN_USE_KQUEUE:
                return False
            try:
                kq = select.kqueue()
            except OSError:
                # likely EMFILE / ENFILE (too many open files)
                return False

            try:
                kev = select.kevent(
                    self.pid,
                    filter=select.KQ_FILTER_PROC,
                    flags=select.KQ_EV_ADD | select.KQ_EV_ONESHOT,
                    fflags=select.KQ_NOTE_EXIT,
                )
                try:
                    events = kq.control([kev], 1, timeout)  # wait
                except OSError:
                    return False
                else:
                    if not events:
                        raise TimeoutExpired(self.args, timeout)
                    return True
            finally:
                kq.close()

        def _wait(self, timeout):
            """Internal implementation of wait() on POSIX.