gh-140137: Handle empty collections in profiling.sampling

Author: pablogsal

Lib/profiling/sampling/sample.py

@@ -642,9 +642,14 @@ def sample(
 
     if output_format == "pstats" and not filename:
         stats = pstats.SampledStats(collector).strip_dirs()
-        print_sampled_stats(
-            stats, sort, limit, show_summary, sample_interval_usec
-        )
+        if not stats.stats:
+            print("No samples were collected.")
+            if mode == PROFILING_MODE_CPU:
+                print("This can happen in CPU mode when all threads are idle.")
+        else:
+            print_sampled_stats(
+                stats, sort, limit, show_summary, sample_interval_usec
+            )
     else:
         collector.export(filename)
 

Lib/pstats.py

@@ -154,6 +154,7 @@ def load_stats(self, arg):
             arg.create_stats()
             self.stats = arg.stats
             arg.stats = {}
+            return
         if not self.stats:
             raise TypeError("Cannot create or construct a %r object from %r"
                             % (self.__class__, arg))

Lib/test/test_profiling/test_sampling_profiler.py

@@ -2644,20 +2644,29 @@ def test_cpu_mode_integration_filtering(self):
 import time
 import threading
 
+# Event to signal when CPU thread is running
+cpu_ready = threading.Event()
+
 def idle_worker():
     time.sleep(999999)
 
 def cpu_active_worker():
+    cpu_ready.set()  # Signal that we're running
     x = 1
     while True:
         x += 1
 
 def main():
-# Start both threads
+    from test.support import SHORT_TIMEOUT
+    # Start both threads
     idle_thread = threading.Thread(target=idle_worker)
     cpu_thread = threading.Thread(target=cpu_active_worker)
     idle_thread.start()
     cpu_thread.start()
+
+    # Wait for CPU thread to be running before continuing
+    cpu_ready.wait(timeout=SHORT_TIMEOUT)
+
     idle_thread.join()
     cpu_thread.join()
 
@@ -2716,6 +2725,37 @@ def main():
             self.assertIn("cpu_active_worker", wall_mode_output)
             self.assertIn("idle_worker", wall_mode_output)
 
+    def test_cpu_mode_with_no_samples(self):
+        """Test that CPU mode handles no samples gracefully when no samples are collected."""
+        # Mock a collector that returns empty stats
+        mock_collector = mock.MagicMock()
+        mock_collector.stats = {}
+        mock_collector.create_stats = mock.MagicMock()
+
+        with (
+            io.StringIO() as captured_output,
+            mock.patch("sys.stdout", captured_output),
+            mock.patch("profiling.sampling.sample.PstatsCollector", return_value=mock_collector),
+            mock.patch("profiling.sampling.sample.SampleProfiler") as mock_profiler_class,
+        ):
+            mock_profiler = mock.MagicMock()
+            mock_profiler_class.return_value = mock_profiler
+
+            profiling.sampling.sample.sample(
+                12345,  # dummy PID
+                duration_sec=0.5,
+                sample_interval_usec=5000,
+                mode=1,  # CPU mode
+                show_summary=False,
+                all_threads=True,
+            )
+
+            output = captured_output.getvalue()
+
+        # Should see the "No samples were collected" message
+        self.assertIn("No samples were collected", output)
+        self.assertIn("CPU mode", output)
+
 
 class TestGilModeFiltering(unittest.TestCase):
     """Test GIL mode filtering functionality (--mode=gil)."""
@@ -2852,20 +2892,29 @@ def test_gil_mode_integration_behavior(self):
 import time
 import threading
 
+# Event to signal when thread is running
+gil_ready = threading.Event()
+
 def gil_releasing_work():
     time.sleep(999999)
 
 def gil_holding_work():
+    gil_ready.set()  # Signal that we're running
     x = 1
     while True:
         x += 1
 
 def main():
-# Start both threads
+    from test.support import SHORT_TIMEOUT
+    # Start both threads
     idle_thread = threading.Thread(target=gil_releasing_work)
     cpu_thread = threading.Thread(target=gil_holding_work)
     idle_thread.start()
     cpu_thread.start()
+
+    # Wait for GIL-holding thread to be running before continuing
+    gil_ready.wait(timeout=SHORT_TIMEOUT)
+
     idle_thread.join()
     cpu_thread.join()