From f56bfc9a5cae4451c9248e73a784e36331e0610c Mon Sep 17 00:00:00 2001 From: "Jeong, YunWon" <69878+youknowone@users.noreply.github.com> Date: Tue, 22 Sep 2026 14:53:52 +0900 Subject: [PATCH] Show stdin traceback lines; omit StopIteration None args (#8778) * Show stdin traceback lines; omit StopIteration None args Register interactive 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 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 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 | 2 - Lib/test/test_pdb.py | 17 ++-- crates/host_env/src/select.rs | 54 +++++++++-- crates/stdlib/src/_tokenize.rs | 7 ++ crates/stdlib/src/socket.rs | 9 +- crates/vm/src/stdlib/sys/monitoring.rs | 6 +- crates/vm/src/vm/interpreter.rs | 118 +++++++++++++++++++++++++ crates/vm/src/vm/mod.rs | 32 ++++--- crates/vm/src/vm/python_run.rs | 9 +- crates/vm/src/vm/vm_new.rs | 7 +- src/shell.rs | 1 + 11 files changed, 214 insertions(+), 48 deletions(-) diff --git a/Lib/test/test_cmd_line_script.py b/Lib/test/test_cmd_line_script.py index a6310738f..8dd64afcc 100644 --- a/Lib/test/test_cmd_line_script.py +++ b/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; traceback omits the source line and caret") def test_repl_stderr_flush(self): self.check_repl_stderr_flush() - @unittest.skip("TODO: RUSTPYTHON; traceback omits the source line and caret") def test_repl_stderr_flush_separate_stderr(self): self.check_repl_stderr_flush(True) diff --git a/Lib/test/test_pdb.py b/Lib/test/test_pdb.py index 25fbf3a9b..471623720 100644 --- a/Lib/test/test_pdb.py +++ b/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')) diff --git a/crates/host_env/src/select.rs b/crates/host_env/src/select.rs index 9dc8b4283..287f64b56 100644 --- a/crates/host_env/src/select.rs +++ b/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() { diff --git a/crates/stdlib/src/_tokenize.rs b/crates/stdlib/src/_tokenize.rs index c071a2b62..b439ddb6e 100644 --- a/crates/stdlib/src/_tokenize.rs +++ b/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(); diff --git a/crates/stdlib/src/socket.rs b/crates/stdlib/src/socket.rs index 232621461..26f6e96ad 100644 --- a/crates/stdlib/src/socket.rs +++ b/crates/stdlib/src/socket.rs @@ -1034,11 +1034,16 @@ mod _socket { where F: FnMut() -> io::Result, { - 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)?; } diff --git a/crates/vm/src/stdlib/sys/monitoring.rs b/crates/vm/src/stdlib/sys/monitoring.rs index 43c797e22..9b32df3a4 100644 --- a/crates/vm/src/stdlib/sys/monitoring.rs +++ b/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, diff --git a/crates/vm/src/vm/interpreter.rs b/crates/vm/src/vm/interpreter.rs index 67f7a0fde..b118af1ce 100644 --- a/crates/vm/src/vm/interpreter.rs +++ b/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, "") + .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 diff --git a/crates/vm/src/vm/mod.rs b/crates/vm/src/vm/mod.rs index 1a8ed3a85..431ce48b6 100644 --- a/crates/vm/src/vm/mod.rs +++ b/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() diff --git a/crates/vm/src/vm/python_run.rs b/crates/vm/src/vm/python_run.rs index f2e1ce0c1..b8b83a240 100644 --- a/crates/vm/src/vm/python_run.rs +++ b/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, source: &str) -> PyResult<()> { + pub fn register_code_in_linecache(&self, code: &Py, 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(()) } diff --git a/crates/vm/src/vm/vm_new.rs b/crates/vm/src/vm/vm_new.rs index 2dd6153d3..0d89fe526 100644 --- a/crates/vm/src/vm/vm_new.rs +++ b/crates/vm/src/vm/vm_new.rs @@ -1060,7 +1060,12 @@ impl VirtualMachine { } pub fn new_stop_iteration(&self, value: Option) -> 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::::new().into(), + }; self.new_payload_exception::( self.ctx.exceptions.stop_iteration.to_owned(), args, diff --git a/src/shell.rs b/src/shell.rs index 055a009ea..8e6d6a069 100644 --- a/src/shell.rs +++ b/src/shell.rs @@ -42,6 +42,7 @@ fn shell_exec( match vm.compile_with_opts(source, compiler::Mode::Single, "", 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) {