Retry select/poll/fcntl on EINTR and add kqueue (#8721)

* Retry select/poll/fcntl on EINTR

select() no longer shares one empty list for r/w/x and now
calls select(2) even with empty sets so the timeout elapses.
poll() rejects type instantiation, treats float timeout as
milliseconds, and raises RuntimeError on concurrent poll().
fcntl, ioctl, flock, and lockf retry after EINTR.

Assisted-by: Grok:grok-4.6

* Add select.kqueue and select.kevent

Expose kqueue(2)/kevent(2) on BSD and macOS, including
control() EINTR retry and marking open kqueue objects
closed after fork.

Assisted-by: Grok:grok-4.6

* Add login_tty and fix termios/syslog/mmap

termios.error is Exception, not OSError.
os.login_tty prepares a tty for a new session.
syslog.openlog encodes ident as UTF-8.
mmap.seek returns the new position.

Assisted-by: Grok:grok-4.6

* Honor mmap trackfd on Unix

When trackfd is false, do not keep a duplicated file
descriptor. size() then fails with EBADF and resize()
raises ValueError.

Assisted-by: Grok:grok-4.6

* address review: portable kevent and kqueue close

Build kevent values by field assignment so FreeBSD's
ext array compiles. Normalize timespec nanoseconds,
keep the kqueue fd live across control(), reinit the
open list lock after fork, make trackfd keyword-only,
and skip login_tty on Solaris/illumos.

Assisted-by: Grok:grok-4.6

Lib/test/_test_eintr.py

@@ -452,7 +452,6 @@ class SignalEINTRTest(EINTRBaseTest):
 class SelectEINTRTest(EINTRBaseTest):
     """ EINTR tests for the select module. """
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_select(self):
         t0 = time.monotonic()
         select.select([], [], [], self.sleep_time)
@@ -460,7 +459,6 @@ class SelectEINTRTest(EINTRBaseTest):
         self.stop_alarm()
         self.check_elapsed_time(dt)
 
-    @unittest.skip("TODO: RUSTPYTHON; timed out at the 10 minute mark")
     @unittest.skipIf(sys.platform == "darwin",
                      "poll may fail on macOS; see issue #28087")
     @unittest.skipUnless(hasattr(select, 'poll'), 'need select.poll')
@@ -544,12 +542,10 @@ class FCNTLEINTRTest(EINTRBaseTest):
 
     # Issue 35633: See https://bugs.python.org/issue35633#msg333662
     # skip test rather than accept PermissionError from all platforms
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; InterruptedError: [Errno 4] Interrupted system call
     @unittest.skipIf(platform.system() == "AIX", "AIX returns PermissionError")
     def test_lockf(self):
         self._lock(fcntl.lockf, "lockf")
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; InterruptedError: [Errno 4] Interrupted system call
     def test_flock(self):
         self._lock(fcntl.flock, "flock")
 

Lib/test/test_mmap.py

@@ -46,7 +46,6 @@ class MmapTests(unittest.TestCase):
         except OSError:
             pass
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; AttributeError: 'mmap' object has no attribute 'seekable'
     def test_basic(self):
         # Test mmap module on Unix systems and Windows
 
@@ -272,7 +271,6 @@ class MmapTests(unittest.TestCase):
                     self.assertRaises(TypeError, m.write_byte, 0)
                     m.close()
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     @unittest.skipIf(os.name == 'nt', 'trackfd not present on Windows')
     def test_trackfd_parameter(self):
         size = 64
@@ -308,7 +306,6 @@ class MmapTests(unittest.TestCase):
                 self.assertEqual(m.closed, True)
                 self.assertEqual(os.stat(TESTFN).st_size, size)
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     @unittest.skipIf(os.name == 'nt', 'trackfd not present on Windows')
     def test_trackfd_neg1(self):
         size = 64

Lib/test/test_poll.py

@@ -185,7 +185,6 @@ class PollTests(unittest.TestCase):
         self.assertRaises(OverflowError, pollster.poll, INT_MAX + 1)
         self.assertRaises(OverflowError, pollster.poll, UINT_MAX + 1)
 
-    @unittest.skip("TODO: RUSTPYTHON; fd reallocation")
     @threading_helper.reap_threads
     def test_threaded_poll(self):
         r, w = os.pipe()

Lib/test/test_pty.py

@@ -89,7 +89,6 @@ def write_all(fd, data):
 
 # Marginal testing of pty suite. Cannot do extensive 'do or fail' testing
 # because pty code is not too portable.
-@unittest.skipIf(getattr(tty, "tcgetwinsize", None) is None, "TODO: RUSTPYTHON; `tty.tcgetwinsize` is required for setUp")
 class PtyTest(unittest.TestCase):
     def setUp(self):
         old_sighup = signal.signal(signal.SIGHUP, self.handle_sighup)
@@ -195,11 +194,6 @@ class PtyTest(unittest.TestCase):
         s2 = _readline(master_fd)
         self.assertEqual(b'For my pet fish, Eric.\n', normalize_output(s2))
 
-    # skip (not expectedFailure) because the test still forks a real child
-    # process, which crashes on missing os.login_tty() inside the parallel
-    # test runner's worker process, corrupting its JSON reporting channel
-    # ("worker bug", reproducible under --slow-ci -j N; not under plain -m test).
-    @unittest.skip("TODO: RUSTPYTHON; pty.fork() calls os.login_tty(), which is not implemented")
     def test_fork(self):
         debug("calling pty.fork()")
         pid, master_fd = pty.fork()
@@ -301,11 +295,6 @@ class PtyTest(unittest.TestCase):
 
         self.assertEqual(data, b"")
 
-    # skip (not expectedFailure) because the test still forks a real child
-    # process, which crashes on missing os.login_tty() inside the parallel
-    # test runner's worker process, corrupting its JSON reporting channel
-    # ("worker bug", reproducible under --slow-ci -j N; not under plain -m test).
-    @unittest.skip("TODO: RUSTPYTHON; pty.fork() calls os.login_tty(), which is not implemented")
     def test_spawn_doesnt_hang(self):
         # gh-140482: Do the test in a pty.fork() child to avoid messing
         # with the interactive test runner's terminal settings.

Lib/test/test_select.py

@@ -82,7 +82,6 @@ class SelectTestCase(unittest.TestCase):
     @unittest.skipIf(
         support.is_emscripten, "Emscripten cannot select a fd multiple times."
     )
-    @unittest.skip("TODO: RUSTPYTHON timed out")
     def test_select_mutated(self):
         a = []
         class F:
@@ -92,7 +91,6 @@ class SelectTestCase(unittest.TestCase):
         a[:] = [F()] * 10
         self.assertEqual(select.select([], a, []), ([], a[:5], []))
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; AssertionError: TypeError not raised by poll
     def test_disallow_instantiation(self):
         support.check_disallow_instantiation(self, type(select.poll()))
 

Lib/test/test_syslog.py

@@ -16,7 +16,6 @@ class Test(unittest.TestCase):
     def tearDown(self):
         syslog.closelog()
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_openlog(self):
         syslog.openlog('python')
         # Issue #6697.

Lib/test/test_termios.py

@@ -289,7 +289,6 @@ class TestModule(unittest.TestCase):
             with self.subTest(name=name):
                 self.assertGreaterEqual(value, 0)
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; AssertionError: <class 'termios.error'> is a subclass of <class 'OSError'>
     def test_exception(self):
         self.assertIsSubclass(termios.error, Exception)
         self.assertNotIsSubclass(termios.error, OSError)

crates/host_env/src/posix.rs

@@ -1013,6 +1013,16 @@ pub fn setresuid(ruid: u32, euid: u32, suid: u32) -> std::io::Result<()> {
         .map_err(std::io::Error::from)
 }
 
+#[cfg(not(any(target_os = "wasi", target_os = "solaris", target_os = "illumos")))]
+pub fn login_tty(fd: i32) -> std::io::Result<()> {
+    let ret = unsafe { libc::login_tty(fd) };
+    if ret < 0 {
+        Err(std::io::Error::last_os_error())
+    } else {
+        Ok(())
+    }
+}
+
 #[cfg(not(target_os = "redox"))]
 pub fn openpty() -> std::io::Result<(OwnedFd, OwnedFd)> {
     let pty = nix::pty::openpty(None, None).map_err(std::io::Error::from)?;

crates/host_env/src/select.rs

@@ -330,3 +330,153 @@ pub mod epoll {
         }
     }
 }
+
+#[cfg(any(
+    target_os = "macos",
+    target_os = "ios",
+    target_os = "freebsd",
+    target_os = "netbsd",
+    target_os = "openbsd",
+    target_os = "dragonfly",
+))]
+pub mod kqueue {
+    use alloc::sync::{Arc, Weak};
+    use core::sync::atomic::{AtomicI32, Ordering};
+    use parking_lot::Mutex;
+    use std::io;
+    use std::os::fd::BorrowedFd;
+
+    pub use libc::{
+        EV_ADD, EV_CLEAR, EV_DELETE, EV_DISABLE, EV_ENABLE, EV_EOF, EV_ERROR, EV_FLAG1, EV_ONESHOT,
+        EV_SYSFLAGS, EVFILT_AIO, EVFILT_PROC, EVFILT_READ, EVFILT_SIGNAL, EVFILT_TIMER,
+        EVFILT_VNODE, EVFILT_WRITE, NOTE_ATTRIB, NOTE_CHILD, NOTE_DELETE, NOTE_EXEC, NOTE_EXIT,
+        NOTE_EXTEND, NOTE_FORK, NOTE_LINK, NOTE_LOWAT, NOTE_PCTRLMASK, NOTE_PDATAMASK, NOTE_RENAME,
+        NOTE_REVOKE, NOTE_TRACK, NOTE_TRACKERR, NOTE_WRITE,
+    };
+
+    #[derive(Copy, Clone, Debug)]
+    pub struct Event {
+        pub ident: usize,
+        pub filter: i16,
+        pub flags: u16,
+        pub fflags: u32,
+        pub data: isize,
+        pub udata: usize,
+    }
+
+    impl Event {
+        pub fn to_libc(self) -> libc::kevent {
+            // Field widths and optional `ext` differ across BSDs; assign
+            // rather than using a struct literal.
+            let mut ev: libc::kevent = unsafe { core::mem::zeroed() };
+            ev.ident = self.ident as _;
+            ev.filter = self.filter as _;
+            ev.flags = self.flags as _;
+            ev.fflags = self.fflags as _;
+            ev.data = self.data as _;
+            ev.udata = self.udata as *mut libc::c_void;
+            ev
+        }
+
+        #[allow(clippy::unnecessary_cast)]
+        pub fn from_libc(e: libc::kevent) -> Self {
+            Self {
+                ident: e.ident as usize,
+                filter: e.filter as i16,
+                flags: e.flags as u16,
+                fflags: e.fflags as u32,
+                data: e.data as isize,
+                udata: e.udata as usize,
+            }
+        }
+    }
+
+    static OPEN: Mutex<Vec<Weak<AtomicI32>>> = Mutex::new(Vec::new());
+
+    fn register_open(cell: &Arc<AtomicI32>) {
+        let mut open = OPEN.lock();
+        open.retain(|w| w.strong_count() > 0);
+        open.push(Arc::downgrade(cell));
+    }
+
+    pub fn create() -> io::Result<Arc<AtomicI32>> {
+        let fd = unsafe { libc::kqueue() };
+        if fd < 0 {
+            return Err(io::Error::last_os_error());
+        }
+        let borrowed = unsafe { BorrowedFd::borrow_raw(fd) };
+        if let Err(err) = crate::posix::set_inheritable(borrowed, false) {
+            let _ = unsafe { libc::close(fd) };
+            return Err(err);
+        }
+        let cell = Arc::new(AtomicI32::new(fd));
+        register_open(&cell);
+        Ok(cell)
+    }
+
+    pub fn from_fd(fd: i32) -> Arc<AtomicI32> {
+        let cell = Arc::new(AtomicI32::new(fd));
+        register_open(&cell);
+        cell
+    }
+
+    pub fn close(cell: &AtomicI32) -> io::Result<()> {
+        let fd = cell.swap(-1, Ordering::SeqCst);
+        if fd < 0 {
+            return Ok(());
+        }
+        let ret = unsafe { libc::close(fd) };
+        if ret < 0 {
+            Err(io::Error::last_os_error())
+        } else {
+            Ok(())
+        }
+    }
+
+    pub fn fd(cell: &AtomicI32) -> i32 {
+        cell.load(Ordering::SeqCst)
+    }
+
+    pub fn kevent(
+        kq: i32,
+        changelist: &[Event],
+        eventlist: &mut [Event],
+        timeout: Option<&libc::timespec>,
+    ) -> io::Result<usize> {
+        let chl: Vec<libc::kevent> = changelist.iter().copied().map(Event::to_libc).collect();
+        let mut evl = vec![unsafe { core::mem::zeroed() }; eventlist.len()];
+        let timeout = timeout.map_or(core::ptr::null(), |t| t);
+        let ret = unsafe {
+            libc::kevent(
+                kq,
+                chl.as_ptr(),
+                chl.len() as _,
+                evl.as_mut_ptr(),
+                evl.len() as _,
+                timeout,
+            )
+        };
+        if ret < 0 {
+            return Err(io::Error::last_os_error());
+        }
+        let n = ret as usize;
+        for (dst, src) in eventlist.iter_mut().zip(evl.into_iter().take(n)) {
+            *dst = Event::from_libc(src);
+        }
+        Ok(n)
+    }
+
+    pub fn mark_closed_after_fork() {
+        // After fork only this thread exists. If the parent held OPEN,
+        // the child's copy stays locked until it is released.
+        if OPEN.try_lock().is_none() {
+            unsafe { OPEN.force_unlock() };
+        }
+        let mut open = OPEN.lock();
+        for weak in open.drain(..) {
+            if let Some(cell) = weak.upgrade() {
+                cell.store(-1, Ordering::SeqCst);
+            }
+        }
+    }
+}

crates/stdlib/src/fcntl.rs

@@ -13,6 +13,21 @@ mod fcntl {
         function::{ArgMemoryBuffer, ArgStrOrBytesLike, Either, OptionalArg},
         stdlib::_io,
     };
+    use rustpython_host_env::io::is_interrupted_error;
+
+    fn retry_on_eintr<T, E: ToPyException>(
+        vm: &VirtualMachine,
+        mut call: impl FnMut() -> Result<T, E>,
+        interrupted: impl Fn(&E) -> bool,
+    ) -> PyResult<T> {
+        loop {
+            match vm.allow_threads(&mut call) {
+                Ok(val) => return Ok(val),
+                Err(err) if interrupted(&err) => vm.check_signals()?,
+                Err(err) => return Err(err.to_pyexception(vm)),
+            }
+        }
+    }
 
     // TODO: supply these from <asm-generic/fnctl.h> (please file an issue/PR upstream):
     //       LOCK_MAND, LOCK_READ, LOCK_WRITE, LOCK_RW, F_GETSIG, F_SETSIG, F_GETLK64, F_SETLK64,
@@ -86,16 +101,21 @@ mod fcntl {
                         .ok_or_else(|| vm.new_value_error("fcntl string arg too long"))?
                         .copy_from_slice(&s)
                 }
-                vm.allow_threads(|| host_fcntl::fcntl_with_bytes(fd, cmd, &mut buf[..arg_len]))
-                    .map_err(|err| err.to_pyexception(vm))?;
+                retry_on_eintr(
+                    vm,
+                    || host_fcntl::fcntl_with_bytes(fd, cmd, &mut buf[..arg_len]),
+                    is_interrupted_error,
+                )?;
                 return Ok(vm.ctx.new_bytes(buf[..arg_len].to_vec()).into());
             }
             OptionalArg::Present(Either::B(i)) => i.as_u32_mask(),
             OptionalArg::Missing => 0,
         };
-        let ret = vm
-            .allow_threads(|| host_fcntl::fcntl_int(fd, cmd, int as i32))
-            .map_err(|err| err.to_pyexception(vm))?;
+        let ret = retry_on_eintr(
+            vm,
+            || host_fcntl::fcntl_int(fd, cmd, int as i32),
+            is_interrupted_error,
+        )?;
         Ok(vm.new_pyobj(ret))
     }
 
@@ -131,11 +151,13 @@ mod fcntl {
                             // copies through one the same way.
                             let mut scratch = vm.new_zeroed_bytes(rw_arg.len())?;
                             scratch.copy_from_slice(&rw_arg.borrow_buf_mut());
-                            let ret = vm
-                                .allow_threads(|| unsafe {
+                            let ret = retry_on_eintr(
+                                vm,
+                                || unsafe {
                                     host_fcntl::ioctl_ptr(fd, request, scratch.as_mut_ptr().cast())
-                                })
-                                .map_err(|err| err.to_pyexception(vm))?;
+                                },
+                                is_interrupted_error,
+                            )?;
                             rw_arg.borrow_buf_mut().copy_from_slice(&scratch);
                             return Ok(vm.ctx.new_int(ret).into());
                         }
@@ -144,16 +166,19 @@ mod fcntl {
                     }
                     Either::B(ro_buf) => fill_buf(&ro_buf.borrow_bytes())?,
                 };
-                vm.allow_threads(|| unsafe {
-                    host_fcntl::ioctl_ptr(fd, request, buf.as_mut_ptr().cast())
-                })
-                .map_err(|err| err.to_pyexception(vm))?;
+                retry_on_eintr(
+                    vm,
+                    || unsafe { host_fcntl::ioctl_ptr(fd, request, buf.as_mut_ptr().cast()) },
+                    is_interrupted_error,
+                )?;
                 Ok(vm.ctx.new_bytes(buf[..buf_len].to_vec()).into())
             }
             Either::B(i) => {
-                let ret = vm
-                    .allow_threads(|| host_fcntl::ioctl_int(fd, request, i))
-                    .map_err(|err| err.to_pyexception(vm))?;
+                let ret = retry_on_eintr(
+                    vm,
+                    || host_fcntl::ioctl_int(fd, request, i),
+                    is_interrupted_error,
+                )?;
                 Ok(vm.ctx.new_int(ret).into())
             }
         }
@@ -165,9 +190,11 @@ mod fcntl {
     fn flock(_io::Fildes(fd): _io::Fildes, operation: i32, vm: &VirtualMachine) -> PyResult {
         // LOCK_EX without LOCK_NB waits for whoever holds the lock, which may
         // be for good.
-        let ret = vm
-            .allow_threads(|| host_fcntl::flock(fd, operation))
-            .map_err(|err| err.to_pyexception(vm))?;
+        let ret = retry_on_eintr(
+            vm,
+            || host_fcntl::flock(fd, operation),
+            is_interrupted_error,
+        )?;
         Ok(vm.ctx.new_int(ret).into())
     }
 
@@ -195,9 +222,14 @@ mod fcntl {
             OptionalArg::Missing => 0,
         };
         // F_LOCK and F_TLOCK differ in exactly this: the first one waits.
-        let ret = vm
-            .allow_threads(|| host_fcntl::lockf(fd, cmd, len, start, whence))
-            .map_err(|err| err.to_pyexception(vm))?;
+        let ret = retry_on_eintr(
+            vm,
+            || host_fcntl::lockf(fd, cmd, len, start, whence),
+            |err| match err {
+                host_fcntl::LockfError::Io(e) => is_interrupted_error(e),
+                _ => false,
+            },
+        )?;
         Ok(vm.ctx.new_int(ret).into())
     }
 }

crates/stdlib/src/mmap.rs

@@ -188,6 +188,8 @@ mod mmap {
         pos: AtomicCell<usize>, // relative to offset
         exports: AtomicCell<usize>,
         access: AccessMode,
+        #[cfg(unix)]
+        trackfd: bool,
     }
 
     impl PyMmap {
@@ -229,6 +231,8 @@ mod mmap {
         access: AccessMode,
         #[pyarg(any, default = 0)]
         offset: i64,
+        #[pyarg(named, default = true)]
+        trackfd: bool,
     }
 
     #[cfg(windows)]
@@ -364,6 +368,7 @@ mod mmap {
                 prot,
                 access,
                 offset,
+                trackfd,
                 ..
             } = args;
 
@@ -443,12 +448,17 @@ mod mmap {
             Ok(Self {
                 closed: AtomicCell::new(false),
                 mmap: PyMutex::new(Some(MmapObj::Mapped(mmap))),
-                fd: AtomicCell::new(fd.map_or(-1, |fd| fd.into_raw())),
+                fd: AtomicCell::new(if trackfd {
+                    fd.map_or(-1, |owned| owned.into_raw())
+                } else {
+                    -1
+                }),
                 offset,
                 size: AtomicCell::new(map_size),
                 pos: AtomicCell::new(0),
                 exports: AtomicCell::new(0),
                 access,
+                trackfd,
             })
         }
 
@@ -752,6 +762,11 @@ mod mmap {
                 return Err(vm.new_buffer_error("mmap can't resize with extant buffers exported."));
             }
 
+            #[cfg(unix)]
+            if !self.trackfd {
+                return Err(vm.new_value_error("mmap can't resize with trackfd=False."));
+            }
+
             if self.access == AccessMode::Write || self.access == AccessMode::Default {
                 return Ok(());
             }
@@ -1095,7 +1110,7 @@ mod mmap {
             dist: isize,
             whence: OptionalArg<core::ffi::c_int>,
             vm: &VirtualMachine,
-        ) -> PyResult<()> {
+        ) -> PyResult<usize> {
             let how = whence.unwrap_or(0);
             let size = self.__len__();
 
@@ -1125,7 +1140,7 @@ mod mmap {
 
             self.pos.store(new_pos as usize);
 
-            Ok(())
+            Ok(new_pos as usize)
         }
 
         #[cfg(unix)]

crates/stdlib/src/select.rs

@@ -116,14 +116,6 @@ mod decl {
         let (wlist, mut w) = seq2set(&wlist)?;
         let (xlist, mut x) = seq2set(&xlist)?;
 
-        if rlist.is_empty() && wlist.is_empty() && xlist.is_empty() {
-            return Ok((
-                vm.ctx.new_list(vec![]),
-                vm.ctx.new_list(vec![]),
-                vm.ctx.new_list(vec![]),
-            ));
-        }
-
         let nfds = cfg_select! {
             windows => 0, // value is ignored on windows
 
@@ -199,7 +191,11 @@ mod decl {
             function::OptionalArg,
             stdlib::_io::Fildes,
         };
-        use core::{convert::TryFrom, time::Duration};
+        use core::{
+            convert::TryFrom,
+            sync::atomic::{AtomicBool, Ordering},
+            time::Duration,
+        };
         use num_traits::{Signed, ToPrimitive};
         use std::time::Instant;
 
@@ -218,7 +214,8 @@ mod decl {
                     if float.is_sign_negative() {
                         None
                     } else {
-                        let secs = if MILLIS { float * 1000.0 } else { float };
+                        // MILLIS: the Python timeout is in milliseconds.
+                        let secs = if MILLIS { float / 1e3 } else { float };
                         Some(Duration::from_secs_f64(secs))
                     }
                 } else if let Some(int) = obj.try_index_opt(vm).transpose()? {
@@ -250,6 +247,7 @@ mod decl {
         pub(crate) struct PyPoll {
             // keep sorted
             fds: PyMutex<Vec<host_select::PollFd>>,
+            poll_running: AtomicBool,
         }
 
         // new EventMask type
@@ -280,7 +278,7 @@ mod decl {
         const DEFAULT_EVENTS: i16 =
             host_select::POLLIN | host_select::POLLPRI | host_select::POLLOUT;
 
-        #[pyclass]
+        #[pyclass(flags(DISALLOW_INSTANTIATION))]
         impl PyPoll {
             #[pymethod]
             fn register(&self, Fildes(fd): Fildes, eventmask: OptionalArg<EventMask>) {
@@ -320,6 +318,17 @@ mod decl {
                 timeout: OptionalArg<TimeoutArg<true>>,
                 vm: &VirtualMachine,
             ) -> PyResult<Vec<PyObjectRef>> {
+                if self.poll_running.swap(true, Ordering::SeqCst) {
+                    return Err(vm.new_runtime_error("concurrent poll() invocation"));
+                }
+                struct ClearRunning<'a>(&'a AtomicBool);
+                impl Drop for ClearRunning<'_> {
+                    fn drop(&mut self) {
+                        self.0.store(false, Ordering::Release);
+                    }
+                }
+                let _running = ClearRunning(&self.poll_running);
+
                 // Poll a copy: the wait releases the GIL-equivalent and runs
                 // signal handlers, which can register or unregister on the same
                 // object, and a held lock would deadlock them.
@@ -569,4 +578,452 @@ mod decl {
             }
         }
     }
+
+    #[cfg(any(
+        target_os = "macos",
+        target_os = "ios",
+        target_os = "freebsd",
+        target_os = "netbsd",
+        target_os = "openbsd",
+        target_os = "dragonfly",
+    ))]
+    #[pyattr(name = "kqueue", once)]
+    fn kqueue_type(_vm: &VirtualMachine) -> PyTypeRef {
+        use crate::vm::class::PyClassImpl;
+        kqueue::PyKqueue::make_static_type()
+    }
+
+    #[cfg(any(
+        target_os = "macos",
+        target_os = "ios",
+        target_os = "freebsd",
+        target_os = "netbsd",
+        target_os = "openbsd",
+        target_os = "dragonfly",
+    ))]
+    #[pyattr(name = "kevent", once)]
+    fn kevent_type(_vm: &VirtualMachine) -> PyTypeRef {
+        use crate::vm::class::PyClassImpl;
+        kqueue::PyKevent::make_static_type()
+    }
+
+    #[cfg(any(
+        target_os = "macos",
+        target_os = "ios",
+        target_os = "freebsd",
+        target_os = "netbsd",
+        target_os = "openbsd",
+        target_os = "dragonfly",
+    ))]
+    #[pyattr]
+    use host_select::kqueue::{
+        EV_ADD as KQ_EV_ADD, EV_CLEAR as KQ_EV_CLEAR, EV_DELETE as KQ_EV_DELETE,
+        EV_DISABLE as KQ_EV_DISABLE, EV_ENABLE as KQ_EV_ENABLE, EV_EOF as KQ_EV_EOF,
+        EV_ERROR as KQ_EV_ERROR, EV_FLAG1 as KQ_EV_FLAG1, EV_ONESHOT as KQ_EV_ONESHOT,
+        EV_SYSFLAGS as KQ_EV_SYSFLAGS, EVFILT_AIO as KQ_FILTER_AIO, EVFILT_PROC as KQ_FILTER_PROC,
+        EVFILT_READ as KQ_FILTER_READ, EVFILT_SIGNAL as KQ_FILTER_SIGNAL,
+        EVFILT_TIMER as KQ_FILTER_TIMER, EVFILT_VNODE as KQ_FILTER_VNODE,
+        EVFILT_WRITE as KQ_FILTER_WRITE, NOTE_ATTRIB as KQ_NOTE_ATTRIB,
+        NOTE_CHILD as KQ_NOTE_CHILD, NOTE_DELETE as KQ_NOTE_DELETE, NOTE_EXEC as KQ_NOTE_EXEC,
+        NOTE_EXIT as KQ_NOTE_EXIT, NOTE_EXTEND as KQ_NOTE_EXTEND, NOTE_FORK as KQ_NOTE_FORK,
+        NOTE_LINK as KQ_NOTE_LINK, NOTE_LOWAT as KQ_NOTE_LOWAT,
+        NOTE_PCTRLMASK as KQ_NOTE_PCTRLMASK, NOTE_PDATAMASK as KQ_NOTE_PDATAMASK,
+        NOTE_RENAME as KQ_NOTE_RENAME, NOTE_REVOKE as KQ_NOTE_REVOKE, NOTE_TRACK as KQ_NOTE_TRACK,
+        NOTE_TRACKERR as KQ_NOTE_TRACKERR, NOTE_WRITE as KQ_NOTE_WRITE,
+    };
+
+    #[cfg(any(
+        target_os = "macos",
+        target_os = "ios",
+        target_os = "freebsd",
+        target_os = "netbsd",
+        target_os = "openbsd",
+        target_os = "dragonfly",
+    ))]
+    pub(super) mod kqueue {
+        use super::*;
+        use crate::vm::{
+            Py, PyObject, PyPayload, PyRef,
+            builtins::{PyFloat, PyType},
+            class_or_notimplemented,
+            common::lock::{PyMutex, PyRwLock},
+            convert::{IntoPyException, ToPyObject},
+            function::{OptionalArg, PyComparisonValue},
+            types::{Comparable, Constructor, Destructor, PyComparisonOp, Representable},
+        };
+        use alloc::sync::Arc;
+        use core::sync::atomic::AtomicI32;
+        use core::time::Duration;
+        use num_traits::ToPrimitive;
+        use std::time::Instant;
+
+        #[pyclass(module = "select", name = "kevent")]
+        #[derive(Debug, PyPayload)]
+        pub(crate) struct PyKevent {
+            ev: PyMutex<host_select::kqueue::Event>,
+        }
+
+        #[derive(FromArgs)]
+        pub(crate) struct KeventNewArgs {
+            #[pyarg(any)]
+            ident: PyObjectRef,
+            #[pyarg(any, default = host_select::kqueue::EVFILT_READ)]
+            filter: i16,
+            #[pyarg(any, default = host_select::kqueue::EV_ADD)]
+            flags: u16,
+            #[pyarg(any, default = 0)]
+            fflags: u32,
+            #[pyarg(any, default = 0)]
+            data: isize,
+            #[pyarg(any, default = 0)]
+            udata: usize,
+        }
+
+        fn ident_from_object(obj: &PyObject, vm: &VirtualMachine) -> PyResult<usize> {
+            if let Some(idx) = obj.try_index_opt(vm) {
+                let n = idx?;
+                n.try_to_primitive(vm).map_err(|_| {
+                    vm.new_overflow_error("Python int too large for C kqueue event identifier")
+                })
+            } else {
+                Selectable::try_from_object(vm, obj.to_owned()).map(|s| s.fno as usize)
+            }
+        }
+
+        impl Constructor for PyKevent {
+            type Args = KeventNewArgs;
+
+            fn py_new(_cls: &Py<PyType>, args: Self::Args, vm: &VirtualMachine) -> PyResult<Self> {
+                let ident = ident_from_object(&args.ident, vm)?;
+                Ok(Self {
+                    ev: PyMutex::new(host_select::kqueue::Event {
+                        ident,
+                        filter: args.filter,
+                        flags: args.flags,
+                        fflags: args.fflags,
+                        data: args.data,
+                        udata: args.udata,
+                    }),
+                })
+            }
+        }
+
+        #[pyclass(with(Constructor, Comparable, Representable))]
+        impl PyKevent {
+            pub(super) fn event(&self) -> host_select::kqueue::Event {
+                *self.ev.lock()
+            }
+
+            pub(super) fn from_event(ev: host_select::kqueue::Event) -> Self {
+                Self {
+                    ev: PyMutex::new(ev),
+                }
+            }
+
+            #[pygetset]
+            fn ident(&self) -> usize {
+                self.ev.lock().ident
+            }
+
+            #[pygetset(setter)]
+            fn set_ident(&self, value: PyObjectRef, vm: &VirtualMachine) -> PyResult<()> {
+                self.ev.lock().ident = ident_from_object(&value, vm)?;
+                Ok(())
+            }
+
+            #[pygetset]
+            fn filter(&self) -> i16 {
+                self.ev.lock().filter
+            }
+
+            #[pygetset(setter)]
+            fn set_filter(&self, value: i16) {
+                self.ev.lock().filter = value;
+            }
+
+            #[pygetset]
+            fn flags(&self) -> u16 {
+                self.ev.lock().flags
+            }
+
+            #[pygetset(setter)]
+            fn set_flags(&self, value: u16) {
+                self.ev.lock().flags = value;
+            }
+
+            #[pygetset]
+            fn fflags(&self) -> u32 {
+                self.ev.lock().fflags
+            }
+
+            #[pygetset(setter)]
+            fn set_fflags(&self, value: u32) {
+                self.ev.lock().fflags = value;
+            }
+
+            #[pygetset]
+            fn data(&self) -> isize {
+                self.ev.lock().data
+            }
+
+            #[pygetset(setter)]
+            fn set_data(&self, value: isize) {
+                self.ev.lock().data = value;
+            }
+
+            #[pygetset]
+            fn udata(&self) -> usize {
+                self.ev.lock().udata
+            }
+
+            #[pygetset(setter)]
+            fn set_udata(&self, value: usize) {
+                self.ev.lock().udata = value;
+            }
+        }
+
+        impl Comparable for PyKevent {
+            fn cmp(
+                zelf: &Py<Self>,
+                other: &PyObject,
+                op: PyComparisonOp,
+                _vm: &VirtualMachine,
+            ) -> PyResult<PyComparisonValue> {
+                let other = class_or_notimplemented!(Self, other);
+                let a = zelf.event();
+                let b = other.event();
+                let ord = a
+                    .ident
+                    .cmp(&b.ident)
+                    .then_with(|| a.filter.cmp(&b.filter))
+                    .then_with(|| a.flags.cmp(&b.flags))
+                    .then_with(|| a.fflags.cmp(&b.fflags))
+                    .then_with(|| a.data.cmp(&b.data))
+                    .then_with(|| a.udata.cmp(&b.udata));
+                Ok(PyComparisonValue::Implemented(op.eval_ord(ord)))
+            }
+        }
+
+        impl Representable for PyKevent {
+            fn repr_str(zelf: &Py<Self>, _vm: &VirtualMachine) -> PyResult<String> {
+                let e = zelf.event();
+                Ok(format!(
+                    "<select.kevent ident={} filter={} flags=0x{:x} fflags=0x{:x} data=0x{:x} udata={:p}>",
+                    e.ident, e.filter, e.flags, e.fflags, e.data, e.udata as *const libc::c_void,
+                ))
+            }
+        }
+
+        #[pyclass(module = "select", name = "kqueue")]
+        #[derive(Debug, PyPayload)]
+        pub(crate) struct PyKqueue {
+            kqfd: PyRwLock<Option<Arc<AtomicI32>>>,
+        }
+
+        impl Constructor for PyKqueue {
+            type Args = ();
+
+            fn py_new(_cls: &Py<PyType>, _args: Self::Args, vm: &VirtualMachine) -> PyResult<Self> {
+                let kqfd = host_select::kqueue::create().map_err(|e| e.into_pyexception(vm))?;
+                Ok(Self {
+                    kqfd: PyRwLock::new(Some(kqfd)),
+                })
+            }
+        }
+
+        impl Destructor for PyKqueue {
+            fn del(zelf: &Py<Self>, _vm: &VirtualMachine) -> PyResult<()> {
+                let cell = zelf.kqfd.write().take();
+                if let Some(cell) = cell {
+                    let _ = host_select::kqueue::close(&cell);
+                }
+                Ok(())
+            }
+        }
+
+        #[derive(FromArgs)]
+        struct KqueueControlArgs {
+            #[pyarg(positional)]
+            changelist: PyObjectRef,
+            #[pyarg(positional)]
+            maxevents: i32,
+            #[pyarg(any, optional)]
+            timeout: OptionalArg<PyObjectRef>,
+        }
+
+        fn timespec_from_timeout(
+            timeout: OptionalArg<PyObjectRef>,
+            vm: &VirtualMachine,
+        ) -> PyResult<Option<libc::timespec>> {
+            let Some(obj) = timeout.into_option() else {
+                return Ok(None);
+            };
+            if vm.is_none(&obj) {
+                return Ok(None);
+            }
+            let secs = if let Some(float) = obj.downcast_ref::<PyFloat>() {
+                float.to_f64()
+            } else if let Some(int) = obj.try_index_opt(vm).transpose()? {
+                int.as_bigint()
+                    .to_f64()
+                    .ok_or_else(|| vm.new_overflow_error("timeout is too large"))?
+            } else {
+                return Err(vm.new_type_error(format!(
+                    "timeout argument must be a number or None, got {}",
+                    obj.class()
+                )));
+            };
+            if secs.is_nan() {
+                return Err(vm.new_value_error("Invalid value NaN (not a number)"));
+            }
+            if secs < 0.0 {
+                return Err(vm.new_value_error("timeout must be positive or None"));
+            }
+            if secs > i64::MAX as f64 {
+                return Err(vm.new_overflow_error("timeout is too large"));
+            }
+            let mut tv_sec = secs.trunc() as i64;
+            let mut tv_nsec = ((secs - tv_sec as f64) * 1e9).round() as i64;
+            if tv_nsec >= 1_000_000_000 {
+                tv_sec = tv_sec.saturating_add(1);
+                tv_nsec -= 1_000_000_000;
+            }
+            Ok(Some(libc::timespec { tv_sec, tv_nsec }))
+        }
+
+        #[pyclass(with(Constructor, Destructor))]
+        impl PyKqueue {
+            #[pymethod]
+            fn close(&self) -> io::Result<()> {
+                let cell = self.kqfd.write().take();
+                if let Some(cell) = cell {
+                    host_select::kqueue::close(&cell)?;
+                }
+                Ok(())
+            }
+
+            #[pygetset]
+            fn closed(&self) -> bool {
+                self.kqfd
+                    .read()
+                    .as_ref()
+                    .is_none_or(|cell| host_select::kqueue::fd(cell) < 0)
+            }
+
+            #[pymethod]
+            fn fileno(&self, vm: &VirtualMachine) -> PyResult<i32> {
+                match self.kqfd.read().as_ref() {
+                    Some(cell) => {
+                        let fd = host_select::kqueue::fd(cell);
+                        if fd < 0 {
+                            Err(vm.new_value_error("I/O operation on closed kqueue object"))
+                        } else {
+                            Ok(fd)
+                        }
+                    }
+                    None => Err(vm.new_value_error("I/O operation on closed kqueue object")),
+                }
+            }
+
+            #[pyclassmethod]
+            fn fromfd(cls: PyTypeRef, fd: i32, vm: &VirtualMachine) -> PyResult<PyRef<Self>> {
+                Self {
+                    kqfd: PyRwLock::new(Some(host_select::kqueue::from_fd(fd))),
+                }
+                .into_ref_with_type(vm, cls)
+            }
+
+            #[pymethod]
+            fn control(&self, args: KqueueControlArgs, vm: &VirtualMachine) -> PyResult<PyListRef> {
+                let guard = self.kqfd.read();
+                let fd = match guard.as_ref() {
+                    Some(cell) => {
+                        let fd = host_select::kqueue::fd(cell);
+                        if fd < 0 {
+                            return Err(vm.new_value_error("I/O operation on closed kqueue object"));
+                        }
+                        fd
+                    }
+                    None => {
+                        return Err(vm.new_value_error("I/O operation on closed kqueue object"));
+                    }
+                };
+                if args.maxevents < 0 {
+                    return Err(vm.new_value_error(format!(
+                        "Length of eventlist must be 0 or positive, got {}",
+                        args.maxevents
+                    )));
+                }
+
+                let mut timeout = timespec_from_timeout(args.timeout, vm)?;
+                let deadline = timeout
+                    .map(|ts| Instant::now() + Duration::new(ts.tv_sec as u64, ts.tv_nsec as u32));
+
+                let changelist = if vm.is_none(&args.changelist) {
+                    Vec::new()
+                } else {
+                    let items: Vec<PyRef<PyKevent>> =
+                        vm.extract_elements_with(&args.changelist, |item| {
+                            item.downcast().map_err(|_| {
+                                vm.new_type_error(
+                                    "changelist must be an iterable of select.kevent objects",
+                                )
+                            })
+                        })?;
+                    if items.len() > i32::MAX as usize {
+                        return Err(vm.new_overflow_error("changelist is too long"));
+                    }
+                    items.iter().map(|ev| ev.event()).collect()
+                };
+
+                let mut eventlist = vec![
+                    host_select::kqueue::Event {
+                        ident: 0,
+                        filter: 0,
+                        flags: 0,
+                        fflags: 0,
+                        data: 0,
+                        udata: 0,
+                    };
+                    args.maxevents as usize
+                ];
+
+                let n = loop {
+                    match vm.allow_threads(|| {
+                        host_select::kqueue::kevent(
+                            fd,
+                            &changelist,
+                            &mut eventlist,
+                            timeout.as_ref(),
+                        )
+                    }) {
+                        Ok(n) => break n,
+                        Err(err) if err.kind() == io::ErrorKind::Interrupted => {
+                            vm.check_signals()?;
+                            if let Some(deadline) = deadline {
+                                if let Some(remaining) =
+                                    deadline.checked_duration_since(Instant::now())
+                                {
+                                    timeout = Some(libc::timespec {
+                                        tv_sec: remaining.as_secs() as i64,
+                                        tv_nsec: remaining.subsec_nanos() as i64,
+                                    });
+                                } else {
+                                    break 0;
+                                }
+                            }
+                        }
+                        Err(err) => return Err(err.into_pyexception(vm)),
+                    }
+                };
+
+                let out = eventlist
+                    .into_iter()
+                    .take(n)
+                    .map(|ev| PyKevent::from_event(ev).to_pyobject(vm))
+                    .collect();
+                Ok(vm.ctx.new_list(out))
+            }
+        }
+    }
 }

crates/stdlib/src/syslog.rs

@@ -7,6 +7,7 @@ mod syslog {
     use crate::vm::{
         PyObjectRef, PyPayload, PyResult, VirtualMachine,
         builtins::{PyStr, PyStrRef},
+        convert::ToPyException,
         function::{OptionalArg, OptionalOption},
         utils::ToCString,
     };
@@ -28,6 +29,27 @@ mod syslog {
     #[pyattr]
     use host_syslog::{LOG_FTP, LOG_INSTALL, LOG_LAUNCHD, LOG_NETINFO, LOG_RAS, LOG_REMOTEAUTH};
 
+    fn ident_to_utf8_cstring(
+        ident: &PyStrRef,
+        vm: &VirtualMachine,
+    ) -> PyResult<alloc::ffi::CString> {
+        let utf8 = ident.to_str().ok_or_else(|| {
+            let start = ident
+                .as_wtf8()
+                .code_points()
+                .position(|c| c.to_char().is_none())
+                .unwrap_or(0);
+            vm.new_unicode_encode_error_real(
+                vm.ctx.new_str("utf-8"),
+                ident.clone(),
+                start,
+                start + 1,
+                vm.ctx.new_str("surrogates not allowed"),
+            )
+        })?;
+        alloc::ffi::CString::new(utf8).map_err(|err| err.to_pyexception(vm))
+    }
+
     fn get_argv(vm: &VirtualMachine) -> Option<PyStrRef> {
         if let Some(argv) = vm.state.config.settings.argv.first()
             && !argv.is_empty()
@@ -58,8 +80,10 @@ mod syslog {
         let logoption = args.logoption.unwrap_or(0);
         let facility = args.facility.unwrap_or(LOG_USER);
         let ident = match args.ident.clone().flatten() {
-            Some(args) => Some(args.to_cstring(vm)?),
-            None => get_argv(vm).map(|argv| argv.to_cstring(vm)).transpose()?,
+            Some(ident) => Some(ident_to_utf8_cstring(&ident, vm)?),
+            None => get_argv(vm)
+                .map(|argv| ident_to_utf8_cstring(&argv, vm))
+                .transpose()?,
         }
         .map(|ident| ident.into_boxed_c_str());
 

crates/stdlib/src/termios.rs

@@ -297,20 +297,17 @@ mod termios {
     }
 
     fn termios_error(err: std::io::Error, vm: &VirtualMachine) -> PyBaseExceptionRef {
-        vm.new_os_subtype_error(
+        vm.new_exception(
             error_type(vm),
-            Some(err.posix_errno()),
-            vm.ctx.new_str(err.to_string()),
+            vec![
+                vm.ctx.new_int(err.posix_errno()).into(),
+                vm.ctx.new_str(err.to_string()).into(),
+            ],
         )
-        .upcast()
     }
 
     #[pyattr(name = "error", once)]
     fn error_type(vm: &VirtualMachine) -> PyTypeRef {
-        vm.ctx.new_exception_type(
-            "termios",
-            "error",
-            Some(vec![vm.ctx.exceptions.os_error.to_owned()]),
-        )
+        vm.ctx.new_exception_type("termios", "error", None)
     }
 }

crates/vm/src/stdlib/posix.rs

@@ -39,7 +39,7 @@ pub mod module {
     use rustpython_host_env::os::ffi::OsStringExt;
     use std::{
         fs, io,
-        os::fd::{BorrowedFd, FromRawFd, IntoRawFd, OwnedFd},
+        os::fd::{AsRawFd, BorrowedFd, FromRawFd, IntoRawFd, OwnedFd},
     };
     use strum::IntoEnumIterator;
     use strum_macros::{EnumIter, EnumString};
@@ -623,6 +623,16 @@ pub mod module {
         crate::signal::clear_after_fork();
         crate::stdlib::_signal::_signal::clear_wakeup_fd_after_fork();
 
+        #[cfg(any(
+            target_os = "macos",
+            target_os = "ios",
+            target_os = "freebsd",
+            target_os = "netbsd",
+            target_os = "openbsd",
+            target_os = "dragonfly",
+        ))]
+        rustpython_host_env::select::kqueue::mark_closed_after_fork();
+
         // Reset weakref stripe locks that may have been held during fork.
         #[cfg(feature = "threading")]
         crate::object::reset_weakref_locks_after_fork();
@@ -1294,6 +1304,13 @@ pub mod module {
             .map_err(|err| err.into_pyexception(vm))
     }
 
+    #[cfg(not(any(target_os = "wasi", target_os = "solaris", target_os = "illumos")))]
+    #[pyfunction]
+    fn login_tty(fd: BorrowedFd<'_>, vm: &VirtualMachine) -> PyResult<()> {
+        rustpython_host_env::posix::login_tty(fd.as_raw_fd())
+            .map_err(|err| err.into_pyexception(vm))
+    }
+
     #[cfg(not(target_os = "redox"))]
     #[pyfunction]
     fn openpty(vm: &VirtualMachine) -> PyResult<(OwnedFd, OwnedFd)> {