Last active
January 27, 2026 03:49
-
-
Save x42005e1f/e2bf1e0352b7cb3ce0c55f354a59394f to your computer and use it in GitHub Desktop.
Yet another microbenchmarking module (draft)
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/usr/bin/env python3 | |
| # SPDX-FileCopyrightText: 2026 Ilya Egorov <[email protected]> | |
| # SPDX-License-Identifier: ISC | |
| # requires-python = ">=3.8" | |
| # dependencies = [ | |
| # "aiologic>=0.15.0", | |
| # "more-itertools>=2.1.0; implementation_name=='cpython'", | |
| # "typing-extensions>=3.10.0; python_version<'3.10'", | |
| # "typing-extensions>=4.6.0; python_version<'3.9'", | |
| # ] | |
| from __future__ import annotations | |
| import sys | |
| import time | |
| from math import inf, isfinite, isinf, isnan | |
| from typing import TYPE_CHECKING | |
| from aiologic.meta import ( | |
| DEFAULT, # aiologic>=0.15.0 | |
| ) | |
| if TYPE_CHECKING: | |
| from typing import Final | |
| from aiologic.meta import ( | |
| DefaultType, # aiologic>=0.15.0 | |
| ) | |
| if sys.implementation.name == "cpython": | |
| from more_itertools import ( | |
| consume, # more-itertools>=2.1.0 | |
| repeatfunc, # more-itertools>=2.1.0 | |
| ) | |
| if TYPE_CHECKING: | |
| if sys.version_info >= (3, 9): # PEP 585 | |
| from collections.abc import Callable, Iterable | |
| else: | |
| from typing import Callable, Iterable | |
| if sys.implementation.name == "cpython": | |
| if sys.version_info >= (3, 10): # python/cpython#23549 | |
| from itertools import pairwise | |
| else: # more-itertools>=2.1.0 | |
| from more_itertools import pairwise | |
| if sys.version_info >= (3, 9): # various bug fixes (caching, etc.) | |
| from typing import Literal | |
| else: # typing-extensions>=4.6.0 | |
| from typing_extensions import Literal | |
| if TYPE_CHECKING: | |
| if sys.version_info >= (3, 10): # PEP 613 | |
| from typing import TypeAlias | |
| else: # typing-extensions>=3.10.0 | |
| from typing_extensions import TypeAlias | |
| ClockName: TypeAlias = Literal[ | |
| "monotonic", | |
| "monotonic_ns", | |
| "perf_counter", | |
| "perf_counter_ns", | |
| "process_time", | |
| "process_time_ns", | |
| "thread_time", | |
| "thread_time_ns", | |
| "time", | |
| "time_ns", | |
| ] | |
| ClockUnit: TypeAlias = Literal[ | |
| "msec", | |
| "nsec", | |
| "sec", | |
| "usec", | |
| ] | |
| _IS_PYPY: Final[bool] = sys.implementation.name == "pypy" | |
| _IS_CPYTHON: Final[bool] = sys.implementation.name == "cpython" | |
| _CLOCK_BY_NAME: Final[dict[ClockName, Callable[[], float]]] = { | |
| "monotonic": time.monotonic, | |
| "monotonic_ns": time.monotonic_ns, | |
| "perf_counter": time.perf_counter, | |
| "perf_counter_ns": time.perf_counter_ns, | |
| "process_time": time.process_time, | |
| "process_time_ns": time.process_time_ns, | |
| **( | |
| { | |
| "thread_time": time.thread_time, | |
| "thread_time_ns": time.thread_time_ns, | |
| } | |
| if hasattr(time, "thread_time") | |
| else {} | |
| ), | |
| "time": time.time, | |
| "time_ns": time.time_ns, | |
| } | |
| _CLOCK_BY_FUNCTION: Final[dict[Callable[[], float], ClockName]] = { | |
| clock_func: clock_name for clock_name, clock_func in _CLOCK_BY_NAME.items() | |
| } | |
| _CLOCK_UNIT_BY_NAME: Final[dict[ClockUnit, float]] = { | |
| "msec": 1e-3, | |
| "nsec": 1e-9, | |
| "sec": 1e-0, | |
| "usec": 1e-6, | |
| } | |
| def measure_one( | |
| impl: Callable[[], Iterable[tuple[float, float]]], | |
| /, | |
| clock_unit: ClockUnit | float, | |
| *, | |
| overhead: float = 0, | |
| timeout: float = 0.2, | |
| ) -> float: | |
| if _IS_PYPY: # avoid bridges (see pypy/pypy#1822) | |
| impl.__code__ = impl.__code__.replace() | |
| if isinstance(clock_unit, str): | |
| try: | |
| clock_unit = _CLOCK_UNIT_BY_NAME[clock_unit] | |
| except KeyError: | |
| msg = "unknown clock unit" | |
| raise ValueError(msg) | |
| if isnan(timeout): | |
| msg = "`timeout` must be non-NaN" | |
| raise ValueError(msg) | |
| if timeout < 0: | |
| msg = "`timeout` must be non-negative" | |
| raise ValueError(msg) | |
| if isinf(timeout): | |
| msg = "`timeout` must be finite" | |
| raise ValueError(msg) | |
| delta = +inf | |
| deadline = -inf | |
| while True: | |
| end = +inf | |
| for start, end in impl(): | |
| assert end >= start, "the clock must not go/jump backwards" | |
| if end != start: | |
| delta = min(end - start, delta) | |
| assert isfinite(end) | |
| if end >= deadline: | |
| if isinf(deadline): | |
| deadline = end + timeout / clock_unit | |
| else: | |
| break | |
| return max(0.0, delta * clock_unit - overhead) | |
| def measure_many( | |
| impl: Callable[[int], Iterable[tuple[float, float]]], | |
| /, | |
| clock_unit: ClockUnit | float, | |
| *, | |
| overhead: float = 0, | |
| timeout: float = 0.2, | |
| ) -> float: | |
| if _IS_PYPY: # avoid bridges (see pypy/pypy#1822) | |
| impl.__code__ = impl.__code__.replace() | |
| if isinstance(clock_unit, str): | |
| try: | |
| clock_unit = _CLOCK_UNIT_BY_NAME[clock_unit] | |
| except KeyError: | |
| msg = "unknown clock unit" | |
| raise ValueError(msg) | |
| if isnan(timeout): | |
| msg = "`timeout` must be non-NaN" | |
| raise ValueError(msg) | |
| if timeout < 0: | |
| msg = "`timeout` must be non-negative" | |
| raise ValueError(msg) | |
| if isinf(timeout): | |
| msg = "`timeout` must be finite" | |
| raise ValueError(msg) | |
| delta = +inf | |
| deadline = -inf | |
| count = next_count = 1 | |
| step = 0 | |
| while True: | |
| current_count = count + step | |
| end = +inf | |
| low = False | |
| for start, end in impl(current_count): | |
| assert end >= start, "the clock must not go/jump backwards" | |
| if end != start: | |
| current_delta = (end - start) / current_count | |
| if overhead: | |
| current_delta -= overhead / (current_count * clock_unit) | |
| if current_delta < delta: | |
| next_count = current_count | |
| delta = current_delta | |
| else: | |
| low = True | |
| assert isfinite(end) | |
| if step > 0: | |
| if (next_step := step << 1) < count or low: | |
| step = next_step | |
| else: | |
| count = next_count | |
| if count != 1: | |
| step = -1 | |
| else: | |
| step = 0 | |
| elif step == 0: | |
| step = +1 | |
| else: | |
| if current_count == 1 or low: | |
| count = next_count | |
| step = 0 | |
| elif -(next_step := step << 1) < count: | |
| step = next_step | |
| else: | |
| step -= current_count >> 1 | |
| if end >= deadline: | |
| if isinf(deadline): | |
| deadline = end + timeout / clock_unit | |
| else: | |
| break | |
| return max(0.0, delta * clock_unit) | |
| def get_clock(name: ClockName) -> Callable[[], float]: | |
| try: | |
| return _CLOCK_BY_NAME[name] | |
| except KeyError: | |
| msg = "unknown clock" | |
| raise ValueError(msg) from None | |
| def get_clock_unit(clock: Callable[[], float] | ClockName) -> ClockUnit: | |
| if isinstance(clock, str): | |
| clock_name = clock | |
| if clock_name not in _CLOCK_BY_NAME: | |
| msg = "unknown clock" | |
| raise ValueError(msg) | |
| else: | |
| try: | |
| clock_name = _CLOCK_BY_FUNCTION[clock] | |
| except KeyError: | |
| msg = "unknown clock" | |
| raise ValueError(msg) from None | |
| if clock_name.endswith("_ns"): | |
| return "nsec" | |
| return "sec" | |
| def get_clock_delta( | |
| clock: Callable[[], float] | ClockName, | |
| clock_unit: ClockUnit | float | DefaultType = DEFAULT, | |
| *, | |
| overhead: float = 0, | |
| timeout: float = 0.2, | |
| ) -> float: | |
| if clock_unit is DEFAULT: | |
| clock_unit = get_clock_unit(clock) | |
| if isinstance(clock, str): | |
| clock = get_clock(clock) | |
| if _IS_CPYTHON: | |
| def impl() -> Iterable[tuple[float, float]]: | |
| return pairwise([*repeatfunc(clock, 4)]) | |
| else: | |
| def impl() -> Iterable[tuple[float, float]]: | |
| a, b, c, d = [clock(), clock(), clock(), clock()] | |
| return [(a, b), (b, c), (c, d)] | |
| return measure_one(impl, clock_unit, overhead=overhead, timeout=timeout) | |
| def get_clock_overhead( | |
| clock: Callable[[], float] | ClockName, | |
| clock_unit: ClockUnit | float | DefaultType = DEFAULT, | |
| *, | |
| overhead: float = 0, | |
| timeout: float = 0.2, | |
| ) -> float: | |
| if clock_unit is DEFAULT: | |
| clock_unit = get_clock_unit(clock) | |
| if isinstance(clock, str): | |
| clock = get_clock(clock) | |
| if _IS_CPYTHON: | |
| def impl(count: int) -> Iterable[tuple[float, float]]: | |
| return pairwise([*repeatfunc(clock, 3 * count + 1)][::count]) | |
| else: | |
| def impl(count: int) -> Iterable[tuple[float, float]]: | |
| a = clock() | |
| b = a | |
| n = count | |
| while n: | |
| n -= 1 | |
| b = clock() | |
| c = b | |
| n = count | |
| while n: | |
| n -= 1 | |
| c = clock() | |
| d = c | |
| n = count | |
| while n: | |
| n -= 1 | |
| d = clock() | |
| return [(a, b), (b, c), (c, d)] | |
| return measure_many(impl, clock_unit, overhead=overhead, timeout=timeout) | |
| def timeit( | |
| func: Callable[[], object], | |
| /, | |
| *, | |
| clock: Callable[[], float] | ClockName, | |
| clock_unit: ClockUnit | float | DefaultType = DEFAULT, | |
| overhead: float = 0, | |
| timeout: float = 0.2, | |
| ) -> float: | |
| if clock_unit is DEFAULT: | |
| clock_unit = get_clock_unit(clock) | |
| if isinstance(clock, str): | |
| clock = get_clock(clock) | |
| if _IS_CPYTHON: | |
| def one() -> Iterable[tuple[float, float]]: | |
| a = clock() | |
| func() | |
| b = clock() | |
| func() | |
| c = clock() | |
| func() | |
| d = clock() | |
| return [(a, b), (b, c), (c, d)] | |
| else: | |
| blackhole = None | |
| def one() -> Iterable[tuple[float, float]]: | |
| nonlocal blackhole # to avoid dead-code elimination | |
| a = clock() | |
| blackhole = func() | |
| b = clock() | |
| blackhole = func() | |
| c = clock() | |
| blackhole = func() | |
| d = clock() | |
| return [(a, b), (b, c), (c, d)] | |
| if _IS_CPYTHON: | |
| def many(count: int) -> Iterable[tuple[float, float]]: | |
| a = clock() | |
| consume(repeatfunc(func, count)) | |
| b = clock() | |
| consume(repeatfunc(func, count)) | |
| c = clock() | |
| consume(repeatfunc(func, count)) | |
| d = clock() | |
| return [(a, b), (b, c), (c, d)] | |
| else: | |
| blackhole = None | |
| def many(count: int) -> Iterable[tuple[float, float]]: | |
| nonlocal blackhole # to avoid dead-code elimination | |
| a = clock() | |
| n = count | |
| while n: | |
| n -= 1 | |
| blackhole = func() | |
| b = clock() | |
| n = count | |
| while n: | |
| n -= 1 | |
| blackhole = func() | |
| c = clock() | |
| n = count | |
| while n: | |
| n -= 1 | |
| blackhole = func() | |
| d = clock() | |
| return [(a, b), (b, c), (c, d)] | |
| return min( | |
| measure_one(one, clock_unit, overhead=overhead, timeout=timeout / 2), | |
| measure_many(many, clock_unit, overhead=overhead, timeout=timeout / 2), | |
| ) | |
| def main() -> None: | |
| perf_counter_times = [] | |
| for clock_name, clock_func in _CLOCK_BY_NAME.items(): | |
| if not clock_name.endswith("_ns"): | |
| info = time.get_clock_info(clock_name) | |
| info_delta = get_clock_delta(clock_name) | |
| info_overhead = get_clock_overhead(clock_name) | |
| if clock_name.startswith("perf_counter"): | |
| perf_counter_times.append((info_delta, info_overhead, clock_name)) | |
| if not clock_name.endswith("_ns"): | |
| print(f"{' ' + clock_name + ' ':=^80}") | |
| else: | |
| print("-" * 80) | |
| print(f"function: {clock_func.__module__}.{clock_func.__qualname__}()") | |
| if not clock_name.endswith("_ns"): | |
| print(f"implementation: {info.implementation}") | |
| print(f"adjustable: {'yes' if info.adjustable else 'no'}") | |
| print(f"monotonic: {'yes' if info.monotonic else 'no'}") | |
| print(f"resolution: {info.resolution}") | |
| print(f"call delta (the smallest measured): {info_delta}") | |
| print(f"call overhead (the smallest measured): {info_overhead}") | |
| print("=" * 80) | |
| clock = min(perf_counter_times)[-1] | |
| def noop() -> None: | |
| pass | |
| for func in ( | |
| object, | |
| type(None), | |
| type(NotImplemented), | |
| type(...), | |
| bool, | |
| int, | |
| float, | |
| complex, | |
| bytes, | |
| str, | |
| tuple, | |
| frozenset, | |
| noop, | |
| ): | |
| func_time = timeit(func, clock=clock, timeout=0.4) | |
| print(f"function: {func.__module__}.{func.__qualname__}()") | |
| print(f"time (the smallest measured): {func_time}") | |
| if __name__ == "__main__": | |
| main() |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment