fix: handle KeyboardInterrupt when sampling and print the result

Signed-off-by: yihong0618 <zouzou0208@gmail.com>

Author: yihong0618

Lib/profiling/sampling/sample.py

@@ -56,50 +56,56 @@ def sample(self, collector, duration_sec=10):
         last_sample_time = start_time
         realtime_update_interval = 1.0  # Update every second
         last_realtime_update = start_time
+        interrupted = False
 
-        while running_time < duration_sec:
-            # Check if live collector wants to stop
-            if hasattr(collector, 'running') and not collector.running:
-                break
-
-            current_time = time.perf_counter()
-            if next_time < current_time:
-                try:
-                    stack_frames = self.unwinder.get_stack_trace()
-                    collector.collect(stack_frames)
-                except ProcessLookupError:
-                    duration_sec = current_time - start_time
+        try:
+            while running_time < duration_sec:
+                # Check if live collector wants to stop
+                if hasattr(collector, 'running') and not collector.running:
                     break
-                except (RuntimeError, UnicodeDecodeError, MemoryError, OSError):
-                    collector.collect_failed_sample()
-                    errors += 1
-                except Exception as e:
-                    if not self._is_process_running():
-                        break
-                    raise e from None
-
-                # Track actual sampling intervals for real-time stats
-                if num_samples > 0:
-                    actual_interval = current_time - last_sample_time
-                    self.sample_intervals.append(
-                        1.0 / actual_interval
-                    )  # Convert to Hz
-                    self.total_samples += 1
-
-                    # Print real-time statistics if enabled
-                    if (
-                        self.realtime_stats
-                        and (current_time - last_realtime_update)
-                        >= realtime_update_interval
-                    ):
-                        self._print_realtime_stats()
-                        last_realtime_update = current_time
-
-                last_sample_time = current_time
-                num_samples += 1
-                next_time += sample_interval_sec
 
+                current_time = time.perf_counter()
+                if next_time < current_time:
+                    try:
+                        stack_frames = self.unwinder.get_stack_trace()
+                        collector.collect(stack_frames)
+                    except ProcessLookupError:
+                        duration_sec = current_time - start_time
+                        break
+                    except (RuntimeError, UnicodeDecodeError, MemoryError, OSError):
+                        collector.collect_failed_sample()
+                        errors += 1
+                    except Exception as e:
+                        if not self._is_process_running():
+                            break
+                        raise e from None
+
+                    # Track actual sampling intervals for real-time stats
+                    if num_samples > 0:
+                        actual_interval = current_time - last_sample_time
+                        self.sample_intervals.append(
+                            1.0 / actual_interval
+                        )  # Convert to Hz
+                        self.total_samples += 1
+
+                        # Print real-time statistics if enabled
+                        if (
+                            self.realtime_stats
+                            and (current_time - last_realtime_update)
+                            >= realtime_update_interval
+                        ):
+                            self._print_realtime_stats()
+                            last_realtime_update = current_time
+
+                    last_sample_time = current_time
+                    num_samples += 1
+                    next_time += sample_interval_sec
+
+                running_time = time.perf_counter() - start_time
+        except KeyboardInterrupt:
+            interrupted = True
             running_time = time.perf_counter() - start_time
+            print("Interrupted by user.")
 
         # Clear real-time stats line if it was being displayed
         if self.realtime_stats and len(self.sample_intervals) > 0:
@@ -120,7 +126,7 @@ def sample(self, collector, duration_sec=10):
             collector.set_stats(self.sample_interval_usec, running_time, sample_rate, error_rate)
 
         expected_samples = int(duration_sec / sample_interval_sec)
-        if num_samples < expected_samples and not is_live_mode:
+        if num_samples < expected_samples and not is_live_mode and not interrupted:
             print(
                 f"Warning: missed {expected_samples - num_samples} samples "
                 f"from the expected total of {expected_samples} "

Misc/NEWS.d/next/Library/2025-11-27-11-39-50.gh-issue-141999._FKGlu.rst

@@ -0,0 +1 @@
+Fix: handle KeyboardInterrupt when sampling and print the result.