Show stdin traceback lines; omit StopIteration None args (#8778)

* Show stdin traceback lines; omit StopIteration None args

Register interactive <stdin> source in linecache so REPL
tracebacks include the source line and caret. Build
StopIteration with empty args when the value is None, and
wrap monitoring STOP_ITERATION the same way.

Unskip REPL stderr flush tests and pdb Flaky doctests.

Assisted-by: Grok:4.6

* Take &Py<PyCode> in register_code_in_linecache

Assisted-by: Grok:4.6

* Count Unknown closing braces in tokenize EOF check

`$` inside an f-string interpolation is TokenKind::Unknown, so
the matching `}` is Unknown rather than Rbrace and generate_tokens
raised TokenError. Also convert &Py<PyCode> to PyObjectRef
explicitly when registering linecache.

Assisted-by: Grok:4.6

* Unskip test_pdb.PdbTestCase.test_zipimport

The test now passes, so the expectedFailure is an unexpected
success under CI.

Assisted-by: Grok:4.6

* Park STW threads; poll sockets after timeout deadline

Check per-thread stop_requested in eval_breaker_tripped so
workers still park when another interpreter clears STOP_BIT.

Start socket timeout after the fd snapshot, and poll once with
timeout 0 when the deadline has already passed, so a readable
socket is not dropped as timed out.

Assisted-by: Grok:4.6

Lib/test/test_cmd_line_script.py

@@ -214,11 +214,9 @@ class CmdLineTest(unittest.TestCase):
     def test_repl_stdout_flush_separate_stderr(self):
         self.check_repl_stdout_flush(True)
 
-    @unittest.skip("TODO: RUSTPYTHON; <stdin> traceback omits the source line and caret")
     def test_repl_stderr_flush(self):
         self.check_repl_stderr_flush()
 
-    @unittest.skip("TODO: RUSTPYTHON; <stdin> traceback omits the source line and caret")
     def test_repl_stderr_flush_separate_stderr(self):
         self.check_repl_stderr_flush(True)
 

Lib/test/test_pdb.py

@@ -743,7 +743,7 @@ def test_list_commands():
     ...     import pdb; pdb.Pdb(nosigint=True, readrc=False).set_trace()
     ...     ret = test_function_2('baz')
 
-    >>> with PdbTestInput([  # TODO: RUSTPYTHON; Flaky # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE +SKIP
+    >>> with PdbTestInput([  # doctest: +ELLIPSIS, +NORMALIZE_WHITESPACE
     ...     'step',      # go to the test function line
     ...     'list',      # list first function
     ...     'step',      # step into second function
@@ -1867,7 +1867,7 @@ def test_pdb_skip_modules_with_callback():
     ...     import pdb; pdb.Pdb(skip=['module_to_skip*'], nosigint=True, readrc=False).set_trace()
     ...     mod.foo_pony(callback)
 
-    >>> with PdbTestInput([  # TODO: RUSTPYTHON; Flaky # doctest: +SKIP
+    >>> with PdbTestInput([
     ...     'step',
     ...     'step',
     ...     'step',
@@ -2336,7 +2336,7 @@ if not SKIP_CORO_TESTS:
         ...     run_yielding_async_fn(test_main)
         ...     print("finished")
 
-        >>> with PdbTestInput(['step',  # TODO: RUSTPYTHON; Flaky # doctest: +SKIP
+        >>> with PdbTestInput(['step',
         ...                    'step',
         ...                    'step',
         ...                    'next',
@@ -2515,7 +2515,7 @@ if not SKIP_CORO_TESTS:
         ...     run_yielding_async_fn(test_main)
         ...     print("finished")
 
-        >>> with PdbTestInput(['step',  # TODO: RUSTPYTHON; Flaky # doctest: +SKIP
+        >>> with PdbTestInput(['step',
         ...                    'step',
         ...                    'step',
         ...                    'next',
@@ -2555,7 +2555,7 @@ def test_pdb_until_command_for_generator():
     ...         print(i)
     ...     print("finished")
 
-    >>> with PdbTestInput(['step',  # TODO: RUSTPYTHON; Flaky # doctest: +SKIP
+    >>> with PdbTestInput(['step',
     ...                    'step',
     ...                    'until 4',
     ...                    'step',
@@ -2612,7 +2612,7 @@ if not SKIP_CORO_TESTS:
         ...     run_yielding_async_fn(test_main)
         ...     print("finished")
 
-        >>> with PdbTestInput(['step',  # TODO: RUSTPYTHON; Flaky # doctest: +SKIP
+        >>> with PdbTestInput(['step',
         ...                    'step',
         ...                    'until 8',
         ...                    'continue']):
@@ -2695,7 +2695,7 @@ def test_pdb_next_command_subiterator():
     ...         print('value', i)
     ...     x = 123
 
-    >>> with PdbTestInput(['step',  # TODO: RUSTPYTHON; Flaky # doctest: +SKIP
+    >>> with PdbTestInput(['step',
     ...                    'step',
     ...                    'step',
     ...                    'next',
@@ -3445,7 +3445,7 @@ def test_pdb_issue_gh_65052():
 
     >>> def test_function():
     ...     A()
-    >>> with PdbTestInput([ # TODO: RUSTPYTHON; Flaky # doctest: +ELLIPSIS +NORMALIZE_WHITESPACE +SKIP
+    >>> with PdbTestInput([ # doctest: +ELLIPSIS +NORMALIZE_WHITESPACE
     ...     's',
     ...     's',
     ...     'retval',
@@ -4614,7 +4614,6 @@ def bœr():
             self.assertIn('42', stdout)
             self.assertIn('return x + 1', stdout)
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     def test_zipimport(self):
         with os_helper.temp_dir() as temp_dir:
             os.mkdir(os.path.join(temp_dir, 'source'))

crates/host_env/src/select.rs

@@ -311,12 +311,16 @@ fn wait_fd_poll(
         let (timeout, is_capped) = match deadline {
             None => (-1, false),
             Some(deadline) => {
-                let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
-                    return Ok(WaitFd::Timeout);
-                };
-                match duration_as_millis_ceiling(remaining) {
-                    Some(ms) => (ms, false),
-                    None => (i32::MAX, true),
+                match deadline.checked_duration_since(Instant::now()) {
+                    // Deadline already passed: still poll once with 0 so a
+                    // readable/writable socket is not reported as timed out
+                    // after scheduling delay between starting the timer and
+                    // entering poll.
+                    None => (0, false),
+                    Some(remaining) => match duration_as_millis_ceiling(remaining) {
+                        Some(ms) => (ms, false),
+                        None => (i32::MAX, true),
+                    },
                 }
             }
         };
@@ -357,9 +361,9 @@ fn wait_fd_select(
     let mut timeout = match deadline {
         None => None,
         Some(deadline) => {
-            let Some(remaining) = deadline.checked_duration_since(Instant::now()) else {
-                return Ok(WaitFd::Timeout);
-            };
+            let remaining = deadline
+                .checked_duration_since(Instant::now())
+                .unwrap_or(Duration::ZERO);
             Some(duration_to_timeval(remaining))
         }
     };
@@ -699,6 +703,38 @@ mod tests {
         assert_eq!(tv.tv_usec as u32, 500_250);
     }
 
+    #[cfg(unix)]
+    #[test]
+    fn wait_fd_sees_ready_socket_after_deadline() {
+        let mut fds = [0; 2];
+        assert_eq!(unsafe { libc::pipe(fds.as_mut_ptr()) }, 0);
+        let (r, w): (i32, i32) = fds.into();
+        let n = unsafe { libc::write(w, b"x".as_ptr().cast(), 1) };
+        assert_eq!(n, 1);
+        let past = Instant::now().checked_sub(Duration::from_secs(1)).unwrap();
+        let result = wait_fd(r, WaitKind::Read, Some(past));
+        unsafe {
+            libc::close(r);
+            libc::close(w);
+        }
+        assert!(matches!(result, Ok(WaitFd::Ready)));
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn wait_fd_times_out_when_deadline_passed_and_not_ready() {
+        let mut fds = [0; 2];
+        assert_eq!(unsafe { libc::pipe(fds.as_mut_ptr()) }, 0);
+        let (r, w): (i32, i32) = fds.into();
+        let past = Instant::now().checked_sub(Duration::from_secs(1)).unwrap();
+        let result = wait_fd(r, WaitKind::Read, Some(past));
+        unsafe {
+            libc::close(r);
+            libc::close(w);
+        }
+        assert!(matches!(result, Ok(WaitFd::Timeout)));
+    }
+
     #[cfg(unix)]
     #[test]
     fn poll_timeout_rounds_fractional_millis_up() {

crates/stdlib/src/_tokenize.rs

@@ -370,6 +370,13 @@ mod _tokenize {
                 .map(|t| match t.kind() {
                     TokenKind::Lpar | TokenKind::Lsqb | TokenKind::Lbrace => 1,
                     TokenKind::Rpar | TokenKind::Rsqb | TokenKind::Rbrace => -1,
+                    // `$` inside an f-string interpolation is Unknown, and the
+                    // matching `}` is then also Unknown rather than Rbrace.
+                    TokenKind::Unknown => match &source[t.range()] {
+                        "(" | "[" | "{" => 1,
+                        ")" | "]" | "}" => -1,
+                        _ => 0,
+                    },
                     _ => 0,
                 })
                 .sum();

crates/stdlib/src/socket.rs

@@ -1034,11 +1034,16 @@ mod _socket {
         where
             F: FnMut() -> io::Result<R>,
         {
-            let deadline = timeout.map(Deadline::new);
+            // Start the deadline after the first snapshot, so lock/scheduling
+            // delay is not subtracted from a short timeout before poll.
+            let mut deadline = None;
 
             loop {
-                if deadline.is_some() || matches!(wait_kind, SockWaitKind::Connect) {
+                if timeout.is_some() || matches!(wait_kind, SockWaitKind::Connect) {
                     let sock = self.sock_snapshot()?;
+                    if deadline.is_none() {
+                        deadline = timeout.map(Deadline::new);
+                    }
                     sock_wait_deadline(&sock, wait_kind, deadline.as_ref(), vm)?;
                 }
 

crates/vm/src/stdlib/sys/monitoring.rs

@@ -1156,11 +1156,7 @@ pub(crate) fn fire_stop_iteration(
     let exc: PyObjectRef = if value.fast_isinstance(vm.ctx.exceptions.stop_iteration) {
         value.clone()
     } else {
-        vm.ctx
-            .exceptions
-            .stop_iteration
-            .as_object()
-            .call(vec![value.clone()], vm)?
+        vm.new_stop_iteration(Some(value.clone())).into()
     };
     fire(
         vm,

crates/vm/src/vm/interpreter.rs

@@ -1733,6 +1733,124 @@ for _ in range(40):
         worker.join().expect("worker panicked");
     }
 
+    /// `STOP_BIT` is process-wide and `start_the_world` in any interpreter
+    /// clears it. A worker must still park from its own `stop_requested`.
+    #[cfg(all(feature = "threading", feature = "rustpython-compiler"))]
+    #[test]
+    fn stop_the_world_parks_when_global_stop_bit_is_cleared() {
+        use crate::compiler::Mode;
+        use alloc::sync::Arc;
+        use core::{
+            sync::atomic::{AtomicBool, AtomicU64, Ordering},
+            time::Duration,
+        };
+
+        let main = Interpreter::without_stdlib(Default::default());
+        let sub = main.create_subinterpreter();
+        let sub_state = sub.enter(|vm| vm.state.clone());
+
+        let progress = Arc::new(AtomicU64::new(0));
+        let stop = Arc::new(AtomicBool::new(false));
+        let flip = Arc::new(AtomicBool::new(true));
+
+        let progress_worker = Arc::clone(&progress);
+        let stop_worker = Arc::clone(&stop);
+        let worker = sub.enter(|vm| {
+            let thread_vm = vm.new_thread();
+            std::thread::spawn(move || {
+                thread_vm.run(|vm| {
+                    let source = "x = 1 + 1\n";
+                    let code = vm
+                        .compile(source, Mode::Exec, "<spin>")
+                        .map_err(|err| err.into_pyexception(vm, Some(source)))
+                        .unwrap();
+                    while !stop_worker.load(Ordering::Acquire) {
+                        let scope = vm.new_scope_with_builtins();
+                        vm.run_code_obj(code.clone(), scope).unwrap();
+                        progress_worker.fetch_add(1, Ordering::Release);
+                    }
+                });
+            })
+        });
+
+        let deadline = std::time::Instant::now() + Duration::from_secs(10);
+        while progress.load(Ordering::Acquire) == 0 {
+            assert!(
+                std::time::Instant::now() < deadline,
+                "worker never started making progress"
+            );
+            std::thread::yield_now();
+        }
+
+        let flip_flag = Arc::clone(&flip);
+        let flicker_cycles = Arc::new(AtomicU64::new(0));
+        let flicker_cycles_thread = Arc::clone(&flicker_cycles);
+        let flicker = std::thread::spawn(move || {
+            while flip_flag.load(Ordering::Acquire) {
+                crate::signal::set_stop_bit();
+                crate::signal::clear_stop_bit();
+                flicker_cycles_thread.fetch_add(1, Ordering::Release);
+            }
+        });
+
+        while flicker_cycles.load(Ordering::Acquire) == 0 {
+            assert!(
+                std::time::Instant::now() < deadline,
+                "STOP_BIT flicker never ran"
+            );
+            std::thread::yield_now();
+        }
+
+        // Stop from a thread of its own so a hang fails the test instead of
+        // stalling the suite. Keep flickering until after the stop completes.
+        let (tx, rx) = std::sync::mpsc::channel();
+        let stopper = std::thread::spawn(move || {
+            sub_state.stop_the_world.stop_the_world(&sub_state);
+            let stopped = tx.send(());
+            (sub_state, stopped)
+        });
+
+        let stopped = rx.recv_timeout(Duration::from_secs(10));
+
+        let parked_at = progress.load(Ordering::Acquire);
+        std::thread::sleep(Duration::from_millis(50));
+        let parked_after = progress.load(Ordering::Acquire);
+
+        // Release before joining: if the stop hung, stopping the worker lets
+        // its slot go away so the stopper can finish.
+        flip.store(false, Ordering::Release);
+        if stopped.is_err() {
+            stop.store(true, Ordering::Release);
+        }
+        flicker.join().expect("stop-bit flicker panicked");
+        let (stop_state, sent) = stopper.join().expect("stopper panicked");
+        if stopped.is_ok() {
+            stop_state.stop_the_world.start_the_world(&stop_state);
+        }
+        sent.expect("send");
+        assert!(
+            stopped.is_ok(),
+            "stop-the-world hung while STOP_BIT was flickering"
+        );
+        assert_eq!(
+            parked_after, parked_at,
+            "subinterpreter thread kept running after STOP_BIT was cleared"
+        );
+
+        let resumed_from = progress.load(Ordering::Acquire);
+        let resume_deadline = std::time::Instant::now() + Duration::from_secs(10);
+        while progress.load(Ordering::Acquire) == resumed_from {
+            assert!(
+                std::time::Instant::now() < resume_deadline,
+                "worker never resumed after start_the_world"
+            );
+            std::thread::yield_now();
+        }
+
+        stop.store(true, Ordering::Release);
+        worker.join().expect("worker panicked");
+    }
+
     /// Entering a subinterpreter from inside the parent's `enter` must attach
     /// the subinterpreter's thread slot (and detach the parent's). Otherwise the
     /// thread runs the sub's bytecode with a DETACHED slot, and a collector

crates/vm/src/vm/mod.rs

@@ -567,11 +567,9 @@ impl StopTheWorldState {
         drop(registry);
         self.thread_countdown.store(0, Ordering::Release);
         self.requester.store(0, Ordering::Relaxed);
-        // Every non-requester thread's `stop_requested` was just cleared
-        // above, so the shared fast-path bit can be cleared too. Safe here
-        // (and only here / `reset_after_fork`) because both run under the
-        // single stop-the-world exclusion, released below, so no new
-        // requester can be setting the bit concurrently.
+        // Drop the process-wide stop hint. Another interpreter may still
+        // have `stop_requested` threads; those keep parking via the
+        // per-thread check in `eval_breaker_tripped`.
         crate::signal::clear_stop_bit();
         #[cfg(debug_assertions)]
         self.debug_assert_all_non_requester_detached(state);
@@ -3500,20 +3498,20 @@ impl VirtualMachine {
         self.get_method(obj, method_name)
     }
 
-    /// Single relaxed load of the shared eval-breaker word, folding in every
-    /// condition that used to require its own check: pending signals, QSBR
-    /// reclamation, scheduled GC, and (under `threading`) finalization and
-    /// stop-the-world. Finalizing sets `FINALIZING_BIT` and any per-thread
-    /// `stop_requested` store also sets `STOP_BIT` (see signal.rs), so this
-    /// no longer needs to read `state.finalizing` or the
-    /// `CURRENT_STOP_REQUESTED` thread-local itself — both are folded into
-    /// the same atomic word `eval_breaker_pending` already loads for
-    /// signals/QSBR/GC. The slow path (`check_signals`) re-derives the exact
-    /// per-thread answer (this thread's own `stop_requested`, `finalizing` +
-    /// `is_main_thread`) once it is taken, so semantics are unchanged; only
-    /// the common (nothing pending) case gets cheaper.
+    /// Fast path for the bytecode loop: pending signals, QSBR, scheduled GC,
+    /// finalization, and stop-the-world.
+    ///
+    /// `STOP_BIT` is a process-wide hint set when any interpreter asks a
+    /// thread to park. Each interpreter's `start_the_world` clears that hint,
+    /// even if another interpreter still has `stop_requested` threads, so the
+    /// per-thread flag is checked first. Missing it lets a worker skip
+    /// `check_signals` and never park, so `stop_the_world` waits forever.
     #[inline]
     pub(crate) fn eval_breaker_tripped(&self) -> bool {
+        #[cfg(feature = "threading")]
+        if thread::stop_requested_for_current_thread() {
+            return true;
+        }
         #[cfg(not(target_arch = "wasm32"))]
         {
             crate::signal::eval_breaker_pending()

crates/vm/src/vm/python_run.rs

@@ -1,7 +1,7 @@
 //! Python code execution functions.
 
 use crate::{
-    AsObject, PyRef, PyResult, VirtualMachine,
+    Py, PyObjectRef, PyResult, VirtualMachine,
     builtins::PyCode,
     compiler::{self},
     scope::Scope,
@@ -30,12 +30,15 @@ impl VirtualMachine {
 
     /// Register a code object's source in linecache._interactive_cache
     /// so that traceback can display source lines and caret indicators.
-    fn register_code_in_linecache(&self, code: &PyRef<PyCode>, source: &str) -> PyResult<()> {
+    pub fn register_code_in_linecache(&self, code: &Py<PyCode>, source: &str) -> PyResult<()> {
         let linecache = self.import("linecache", 0)?;
         let register = linecache.get_attr("_register_code", self)?;
         let source_str = self.ctx.new_str(source);
         let filename = self.ctx.new_str(code.source_path().as_str());
-        register.call((code.as_object().to_owned(), source_str, filename), self)?;
+        register.call(
+            (PyObjectRef::from(code.to_owned()), source_str, filename),
+            self,
+        )?;
         Ok(())
     }
 

crates/vm/src/vm/vm_new.rs

@@ -1060,7 +1060,12 @@ impl VirtualMachine {
     }
 
     pub fn new_stop_iteration(&self, value: Option<PyObjectRef>) -> PyBaseExceptionRef {
-        let args: FuncArgs = value.map(|v| vec![v]).unwrap_or_default().into();
+        // None (or omitted) is PyErr_SetNone: args stay empty so
+        // format_exception_only prints "StopIteration" not "StopIteration: None".
+        let args: FuncArgs = match value {
+            Some(v) if !self.is_none(&v) => vec![v].into(),
+            _ => Vec::<PyObjectRef>::new().into(),
+        };
         self.new_payload_exception::<PyStopIteration>(
             self.ctx.exceptions.stop_iteration.to_owned(),
             args,

src/shell.rs

@@ -42,6 +42,7 @@ fn shell_exec(
     match vm.compile_with_opts(source, compiler::Mode::Single, "<stdin>", opts) {
         Ok(code) => {
             *future_features |= code.code.flags & CodeFlags::FUTURE_MASK;
+            let _ = vm.register_code_in_linecache(&code, source);
             if empty_line_given || !continuing_block {
                 // We want to execute the full code
                 match vm.run_code_obj(code, scope) {