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) {