perf(serverless): defer CPU probe off worker import path

runpod/serverless/utils/rp_debugger.py

@@ -6,20 +6,30 @@
 
 import datetime
 from datetime import timezone
+import functools
 import platform
 import time
 
-import cpuinfo
-
 # ---------------------------------------------------------------------------- #
 #                                  System Info                                 #
 # ---------------------------------------------------------------------------- #
 OS_INFO = f"{platform.system()} {platform.release()}"
 
-try:
-    PROCESSOR = cpuinfo.get_cpu_info()["brand_raw"]
-except KeyError:
-    PROCESSOR = "Unable to get processor info."
+PROCESSOR_UNKNOWN = "Unable to get processor info."
+
+
+@functools.cache
+def get_processor_info():
+    """
+    Return the processor brand name, probing the CPU on first call only.
+
+    The probe takes ~75ms, so it is kept off the worker import path and only
+    runs when a job requests debugger output.
+    """
+    import cpuinfo  # pylint: disable=import-outside-toplevel
+
+    return cpuinfo.get_cpu_info().get("brand_raw", PROCESSOR_UNKNOWN)
+
 
 PYTHON_VERSION = platform.python_version()
 
@@ -192,7 +202,7 @@ def get_debugger_output():
 
     system_info = {
         "os": OS_INFO,
-        "processor": PROCESSOR,
+        "processor": get_processor_info(),
         "python_version": PYTHON_VERSION,
         "runpod": runpod.__version__,
     }

tests/test_performance/test_cold_start.py

@@ -242,6 +242,19 @@ def test_cold_start_benchmark(tmp_path):
     ), "Import time exceeds 2000ms"
 
 
+def test_worker_import_path_skips_cpu_probe():
+    """
+    The worker job module must not load cpuinfo at import time.
+
+    rp_job imports rp_debugger unconditionally; probing the CPU there costs
+    ~75ms on every worker cold start even though the result is only used when
+    a job enables rp_debugger.
+    """
+    assert not check_module_loaded(
+        "import runpod.serverless.modules.rp_job", "cpuinfo"
+    )
+
+
 if __name__ == "__main__":
     results = run_full_benchmark()
     print("\nFull Results:")

tests/test_serverless/test_utils/test_debugger.py

@@ -2,7 +2,6 @@
 Unit tests for the debugger utility functions.
 """
 
-import importlib
 import time
 import unittest
 from unittest.mock import patch
@@ -24,17 +23,46 @@ def setUp(self):
         self.checkpoints = Checkpoints()
         self.checkpoints.clear()
 
-    @patch("runpod.serverless.utils.rp_debugger.cpuinfo.get_cpu_info")
-    def test_key_error(self, mock_get_cpu_info):
+    def tearDown(self):
+        rp_debugger.get_processor_info.cache_clear()
+
+    @patch("cpuinfo.get_cpu_info")
+    def test_processor_info_fallback_on_missing_brand(self, mock_get_cpu_info):
+        """
+        Missing brand_raw falls back to a placeholder instead of raising.
+        """
+        mock_get_cpu_info.return_value = {}
+        rp_debugger.get_processor_info.cache_clear()
+
+        self.assertEqual(
+            rp_debugger.get_processor_info(), "Unable to get processor info."
+        )
+
+    @patch("cpuinfo.get_cpu_info")
+    def test_processor_info_probed_once(self, mock_get_cpu_info):
+        """
+        The CPU probe runs on first use only and is cached afterwards.
+        """
+        mock_get_cpu_info.return_value = {"brand_raw": "Test CPU"}
+        rp_debugger.get_processor_info.cache_clear()
+
+        first = rp_debugger.get_processor_info()
+        second = rp_debugger.get_processor_info()
+
+        self.assertEqual((first, second), ("Test CPU", "Test CPU"))
+        self.assertEqual(mock_get_cpu_info.call_count, 1)
+
+    @patch("cpuinfo.get_cpu_info")
+    def test_debugger_output_includes_processor(self, mock_get_cpu_info):
         """
-        Test that a KeyError is raised when an invalid key is used.
+        Debugger output reports the probed processor name.
         """
-        mock_get_cpu_info.side_effect = KeyError("Test Error")
+        mock_get_cpu_info.return_value = {"brand_raw": "Test CPU"}
+        rp_debugger.get_processor_info.cache_clear()
 
-        importlib.reload(rp_debugger)
+        system_info = get_debugger_output()["system_info"]
 
-        assert mock_get_cpu_info.called
-        self.assertEqual(rp_debugger.PROCESSOR, "Unable to get processor info.")
+        self.assertEqual(system_info["processor"], "Test CPU")
 
     def test_checkpoints(self):
         """