MCPcopy Index your code
hub / github.com/RustPython/RustPython / stop_the_world

Method stop_the_world

crates/vm/src/vm/mod.rs:326–387  ·  view source on GitHub ↗

Stop all non-requester threads (`stop_the_world`). 1. Sets `requested`, marking the requester thread. 2. CAS detached threads to SUSPENDED. 3. Waits (polling with 1 ms condvar timeout) for attached threads to self-suspend in `check_signals`.

(&self, vm: &VirtualMachine)

Source from the content-addressed store, hash-verified

324 /// 3. Waits (polling with 1 ms condvar timeout) for attached threads
325 /// to self-suspend in `check_signals`.
326 pub fn stop_the_world(&self, vm: &VirtualMachine) {
327 let start = std::time::Instant::now();
328 let requester_ident = crate::stdlib::_thread::get_ident();
329 self.requester.store(requester_ident, Ordering::Relaxed);
330 self.stats_stop_calls.fetch_add(1, Ordering::Relaxed);
331 let initial_countdown = self.init_thread_countdown(vm);
332 stw_trace(format_args!("stop begin requester={requester_ident}"));
333 if initial_countdown == 0 {
334 self.world_stopped.store(true, Ordering::Release);
335 #[cfg(debug_assertions)]
336 self.debug_assert_all_non_requester_suspended(vm);
337 stw_trace(format_args!(
338 "stop end requester={requester_ident} wait_ns=0 polls=0"
339 ));
340 return;
341 }
342
343 let mut polls = 0u64;
344 loop {
345 if self.park_detached_threads(vm) {
346 break;
347 }
348 polls = polls.saturating_add(1);
349 // Wait up to 1 ms for a thread to notify us it suspended.
350 // Re-check under the wait mutex first to avoid a lost-wake race:
351 // a thread may have suspended and notified right before we enter wait.
352 let guard = self.notify_mutex.lock().unwrap();
353 if self.thread_countdown.load(Ordering::Acquire) == 0 || self.park_detached_threads(vm)
354 {
355 drop(guard);
356 break;
357 }
358 let _ = self
359 .notify_cv
360 .wait_timeout(guard, core::time::Duration::from_millis(1));
361 }
362 if polls != 0 {
363 self.stats_poll_loops.fetch_add(polls, Ordering::Relaxed);
364 }
365 let wait_ns = start.elapsed().as_nanos().min(u128::from(u64::MAX)) as u64;
366 self.stats_last_wait_ns.store(wait_ns, Ordering::Relaxed);
367 self.stats_total_wait_ns
368 .fetch_add(wait_ns, Ordering::Relaxed);
369 let mut prev_max = self.stats_max_wait_ns.load(Ordering::Relaxed);
370 while wait_ns > prev_max {
371 match self.stats_max_wait_ns.compare_exchange_weak(
372 prev_max,
373 wait_ns,
374 Ordering::Relaxed,
375 Ordering::Relaxed,
376 ) {
377 Ok(_) => break,
378 Err(observed) => prev_max = observed,
379 }
380 }
381 self.world_stopped.store(true, Ordering::Release);
382 #[cfg(debug_assertions)]
383 self.debug_assert_all_non_requester_suspended(vm);

Callers 1

py_os_before_forkFunction · 0.80

Calls 10

stw_traceFunction · 0.85
init_thread_countdownMethod · 0.80
park_detached_threadsMethod · 0.80
get_identFunction · 0.50
storeMethod · 0.45
unwrapMethod · 0.45
lockMethod · 0.45
loadMethod · 0.45
minMethod · 0.45

Tested by

no test coverage detected