diff --git a/CHANGES.md b/CHANGES.md index eb539e0d..8dd14f59 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -1,6 +1,12 @@ 3.9.0 (under development) ========================= +- Report the calling thread's OpenMP `blocktime_ms` in `threadpool_info()` and + controller information using LLVM/Intel OpenMP's `kmp_get_blocktime` when + available. The value is queried dynamically and reported in milliseconds; + other OpenMP runtimes report `None`. + https://github.com/joblib/threadpoolctl/issues/270 + 3.8.0 (2026-11-05) ================== diff --git a/README.md b/README.md index ec393ffa..e46a33e0 100644 --- a/README.md +++ b/README.md @@ -59,7 +59,8 @@ python -m threadpoolctl -i numpy scipy.linalg "user_api": "openmp", "internal_api": "openmp", "version": null, - "num_threads": 4 + "num_threads": 4, + "blocktime_ms": 200 } ] ``` @@ -87,7 +88,8 @@ that are loaded when importing Python packages: 'threading_layer': 'intel', 'user_api': 'blas', 'version': '2019.0.4'}, - {'filepath': '/home/ogrisel/miniconda3/envs/tmp/lib/libiomp5.so', + {'blocktime_ms': 200, + 'filepath': '/home/ogrisel/miniconda3/envs/tmp/lib/libiomp5.so', 'internal_api': 'openmp', 'num_threads': 4, 'prefix': 'libiomp', @@ -103,13 +105,15 @@ that are loaded when importing Python packages: 'threading_layer': 'intel', 'user_api': 'blas', 'version': '2019.0.4'}, - {'filepath': '/home/ogrisel/miniconda3/envs/tmp/lib/libiomp5.so', + {'blocktime_ms': 200, + 'filepath': '/home/ogrisel/miniconda3/envs/tmp/lib/libiomp5.so', 'internal_api': 'openmp', 'num_threads': 4, 'prefix': 'libiomp', 'user_api': 'openmp', 'version': None}, - {'filepath': '/home/ogrisel/miniconda3/envs/tmp/lib/libgomp.so.1.0.0', + {'blocktime_ms': None, + 'filepath': '/home/ogrisel/miniconda3/envs/tmp/lib/libgomp.so.1.0.0', 'internal_api': 'openmp', 'num_threads': 4, 'prefix': 'libgomp', @@ -122,6 +126,17 @@ with MKL and its Intel OpenMP (`libiomp5`) implementation while `xgboost` was in from pypi.org and links against GNU OpenMP (`libgomp`) so both OpenMP runtimes are loaded in the same Python program. +For OpenMP libraries, `blocktime_ms` reports the calling thread's active wait +time before worker threads sleep, in milliseconds. It is `None` when the runtime +does not expose `kmp_get_blocktime` (for example, GNU OpenMP). A valid +`KMP_BLOCKTIME=35001us` startup setting reports `35.001`; microsecond settings +require a runtime that supports them. The startup setting must remain in the +environment until the loaded runtime is first inspected. Its unit is then cached +per runtime, while the thread's value is queried on every call, including after +`kmp_set_blocktime` and when reusing a controller. Different Python threads can +report different values. The native API has no unit getter, so changing units +with `kmp_set_defaults` after initialization is not supported. + The state of these libraries is also accessible through the object oriented API: ```python diff --git a/tests/test_openmp.py b/tests/test_openmp.py new file mode 100644 index 00000000..32fa8c57 --- /dev/null +++ b/tests/test_openmp.py @@ -0,0 +1,163 @@ +"""Introspection of OpenMP runtime settings.""" + +import ctypes +import json +import os +import re +import subprocess +import sys +from concurrent.futures import ThreadPoolExecutor +from types import SimpleNamespace + +import pytest + +from threadpoolctl import ( + _CDLL_CACHE, + OpenMPController, + ThreadpoolController, + threadpool_info, +) + +# Import the OpenMP helpers, if available, to load the runtimes used by CI. +from . import utils as _ # noqa: F401 + + +@pytest.mark.parametrize("debugging_info", [False, True]) +def test_blocktime_missing_symbol(monkeypatch, debugging_info): + dynlib = SimpleNamespace( + omp_get_max_threads=lambda: 2, + omp_set_num_threads=lambda num_threads: None, + ) + monkeypatch.setattr(_CDLL_CACHE, "get_cdll", lambda filepath: dynlib) + controller = OpenMPController(filepath="libgomp.so", prefix="libgomp") + + assert controller.get_blocktime() is None + assert controller.blocktime_ms is None + info = controller.info(debugging_info=debugging_info) + assert info["blocktime_ms"] is None + assert ("thread_limit_scope" in info) == debugging_info + + +@pytest.fixture +def kmp_controller(): + controllers = ThreadpoolController().select(user_api="openmp").lib_controllers + for controller in controllers: + if hasattr(controller.dynlib, "kmp_get_blocktime") and hasattr( + controller.dynlib, "kmp_set_blocktime" + ): + return controller + pytest.skip("Requires an OpenMP runtime with the kmp blocktime API") + + +def test_blocktime_actualized(kmp_controller): + """A reused controller reflects changes made through the native API.""" + get_blocktime = kmp_controller.dynlib.kmp_get_blocktime + get_blocktime.argtypes = [] + get_blocktime.restype = ctypes.c_int + set_blocktime = kmp_controller.dynlib.kmp_set_blocktime + set_blocktime.argtypes = [ctypes.c_int] + set_blocktime.restype = None + controller = ThreadpoolController().select(filepath=kmp_controller.filepath) + original = get_blocktime() + try: + for value in (0, 17): + set_blocktime(value) + assert get_blocktime() == value + assert kmp_controller.get_blocktime() == value + assert kmp_controller.blocktime_ms == value + assert controller.info()[0]["blocktime_ms"] == value + assert controller.select(blocktime_ms=value).info() == controller.info() + info = next( + lib + for lib in threadpool_info() + if lib["filepath"] == kmp_controller.filepath + ) + assert info["blocktime_ms"] == value + finally: + set_blocktime(original) + + +def test_blocktime_current_thread(kmp_controller): + """Reading a shared controller reports the calling thread's blocktime.""" + get_blocktime = kmp_controller.dynlib.kmp_get_blocktime + get_blocktime.argtypes = [] + get_blocktime.restype = ctypes.c_int + set_blocktime = kmp_controller.dynlib.kmp_set_blocktime + set_blocktime.argtypes = [ctypes.c_int] + set_blocktime.restype = None + original = get_blocktime() + + def worker(): + worker_original = get_blocktime() + try: + set_blocktime(23) + return kmp_controller.info()["blocktime_ms"], kmp_controller.blocktime_ms + finally: + set_blocktime(worker_original) + + try: + set_blocktime(7) + assert kmp_controller.info()["blocktime_ms"] == 7 + with ThreadPoolExecutor(max_workers=1) as executor: + assert executor.submit(worker).result() == (23, 23) + assert get_blocktime() == 7 + assert kmp_controller.info()["blocktime_ms"] == 7 + finally: + set_blocktime(original) + + +@pytest.mark.parametrize("setting", [None, "37", "37ms", "35001us", "35 us"]) +def test_blocktime_ms_matches_verbose_runtime_in_subprocess(kmp_controller, setting): + """Compare introspection with the effective setting printed by the runtime.""" + env = os.environ.copy() + env["OMP_DISPLAY_ENV"] = "VERBOSE" + env["KMP_SETTINGS"] = "1" + env.pop("KMP_BLOCKTIME", None) + if setting is not None: + env["KMP_BLOCKTIME"] = setting + code = """ +import ctypes +import json +import os +import sys +from threadpoolctl import threadpool_info + +runtime = ctypes.CDLL(sys.argv[1]) +runtime.omp_get_max_threads() +initial = threadpool_info() +# The native unit stays fixed when the environment changes after initialization. +os.environ["KMP_BLOCKTIME"] = "1us" if os.environ.get("KMP_BLOCKTIME", "").endswith("ms") else "1ms" +runtime.kmp_set_blocktime(17) +print("THREADPOOLCTL_INFO=" + json.dumps({"initial": initial, "updated": threadpool_info()})) +""" + completed = subprocess.run( + [sys.executable, "-c", code, kmp_controller.filepath], + env=env, + capture_output=True, + text=True, + check=True, + timeout=30, + ) + output = completed.stdout + completed.stderr + # Effective settings follow any user settings in KMP_SETTINGS output. + settings = re.findall(r"KMP_BLOCKTIME\s*=\s*['\"]?(\d+)\s*(us|ms)?", output) + assert settings, output + value, unit = settings[-1] + if setting is not None and setting.endswith("us") and unit != "us": + pytest.skip("This runtime does not support microsecond KMP_BLOCKTIME settings") + expected_ms = int(value) / 1000 if unit == "us" else int(value) + match = re.search(r"^THREADPOOLCTL_INFO=(.*)$", completed.stdout, re.MULTILINE) + assert match, output + libraries = json.loads(match.group(1)) + info = next( + lib + for lib in libraries["initial"] + if lib["filepath"] == kmp_controller.filepath + ) + assert info["blocktime_ms"] == pytest.approx(expected_ms) + updated = next( + lib + for lib in libraries["updated"] + if lib["filepath"] == kmp_controller.filepath + ) + assert updated["blocktime_ms"] == pytest.approx(0.017 if unit == "us" else 17) diff --git a/threadpoolctl.py b/threadpoolctl.py index 40377c03..d57df651 100644 --- a/threadpoolctl.py +++ b/threadpoolctl.py @@ -234,7 +234,8 @@ class LibController(ABC): - version : version of the library (if available). In addition, each library controller may expose internal API specific entries. They - must be set as attributes in the `set_additional_attributes` method. + can be set as attributes in the `set_additional_attributes` method, or added + by overriding `info` when their values need to be queried dynamically. """ @final @@ -940,6 +941,37 @@ class OpenMPController(LibController): "omp_get_num_threads", ) + def info(self, debugging_info: bool = False): + return {**super().info(debugging_info), "blocktime_ms": self.blocktime_ms} + + @property + def blocktime_ms(self): + """The calling thread's blocktime in milliseconds, queried dynamically.""" + return self.get_blocktime() + + def get_blocktime(self): + """Return the calling thread's blocktime in ms, or None if unavailable. + + The native API returns the units of its KMP_BLOCKTIME startup setting. + That setting must still be present when the runtime is first inspected. + Changing units with kmp_set_defaults is not supported: the kmp API has + no unit getter. Microsecond settings require a runtime supporting them. + """ + get_func = self._get_symbol("kmp_get_blocktime") + if get_func is None: + return None + get_func.argtypes = [] + get_func.restype = ctypes.c_int + value = get_func() + return value / 1000 if self._blocktime_uses_microseconds() else value + + @_CDLL_CACHE.cache_method_on_dynlib + def _blocktime_uses_microseconds(self): + # Cache the startup unit per loaded runtime, never the thread-local value. + # The runtime's __kmp_blocktime_units is not part of its exported API. + setting = os.environ.get("KMP_BLOCKTIME", "") + return re.fullmatch(r"\s*\d+\s*us\s*", setting, re.IGNORECASE) is not None + def get_num_threads(self): get_func = getattr(self.dynlib, "omp_get_max_threads", lambda: None) return get_func() @@ -1016,6 +1048,11 @@ def threadpool_info(debugging_info: bool = False): - "num_threads": the current thread limit. In addition, each library may contain internal_api specific entries. + OpenMP libraries expose "blocktime_ms", the calling thread's active wait + time in milliseconds, or None when unavailable. LLVM/Intel OpenMP values + are read from ``kmp_get_blocktime`` on every call and normalized using the + valid ``KMP_BLOCKTIME`` startup setting, which must be preserved until + first inspection. Unit changes through ``kmp_set_defaults`` are unsupported. Parameters ----------