Micro-benchmark for hourly cron minute alignment logic

benchmark.py

#!/usr/bin/env python3
"""Micro-benchmark for hourly cron minute alignment logic.

This compares the old minute extraction:
    int((timestamp // 60) % 60)

against the new timezone-aware minute extraction:
    datetime.datetime.fromtimestamp(timestamp, tz=tz).minute

It also benchmarks the full candidate-selection loop used by the hourly
schedule fast path so we can estimate end-to-end overhead more realistically.
"""

from __future__ import annotations

import argparse
import datetime as dt
import math
import statistics
import time
from collections.abc import Callable, Sequence
from zoneinfo import ZoneInfo

SECONDS_PER_MINUTE = 60
MINUTES_PER_HOUR = 60

DEFAULT_TIMEZONES = [
    "UTC",
    "Asia/Kolkata",
    "Asia/Kathmandu",
    "Australia/Adelaide",
    "America/St_Johns",
    "Pacific/Chatham",
    "Europe/Berlin",
    "America/New_York",
]


def _old_minute(timestamp: float, tz: dt.tzinfo) -> int:
    del tz
    return int((timestamp // SECONDS_PER_MINUTE) % MINUTES_PER_HOUR)


def _new_minute(timestamp: float, tz: dt.tzinfo) -> int:
    return dt.datetime.fromtimestamp(timestamp, tz=tz).minute


def _hourly_step(
    timestamp: float,
    minutes: Sequence[int],
    tz: dt.tzinfo,
    ascending: bool,
    minute_getter: Callable[[float, dt.tzinfo], int],
) -> float:
    if ascending:
        new_timestamp = math.ceil(timestamp)
        new_timestamp = (
            new_timestamp
            + (SECONDS_PER_MINUTE - new_timestamp % SECONDS_PER_MINUTE) % SECONDS_PER_MINUTE
        )
        current_minute = minute_getter(new_timestamp, tz)
        final_timestamp = None

        for minute in minutes:
            new_timestamp_cand = new_timestamp + SECONDS_PER_MINUTE * (
                (minute - current_minute) % MINUTES_PER_HOUR
            )
            if new_timestamp_cand <= timestamp:
                new_timestamp_cand += SECONDS_PER_MINUTE * MINUTES_PER_HOUR
            final_timestamp = (
                new_timestamp_cand
                if final_timestamp is None
                else min(final_timestamp, new_timestamp_cand)
            )
    else:
        new_timestamp = math.floor(timestamp)
        new_timestamp -= new_timestamp % SECONDS_PER_MINUTE
        current_minute = minute_getter(new_timestamp, tz)
        final_timestamp = None

        for minute in minutes:
            new_timestamp_cand = new_timestamp - SECONDS_PER_MINUTE * (
                (current_minute - minute) % MINUTES_PER_HOUR
            )
            if new_timestamp_cand >= timestamp:
                new_timestamp_cand -= SECONDS_PER_MINUTE * MINUTES_PER_HOUR
            final_timestamp = (
                new_timestamp_cand
                if final_timestamp is None
                else max(final_timestamp, new_timestamp_cand)
            )

    return float(final_timestamp)


def _build_timestamps(tz: dt.tzinfo, samples: int) -> list[float]:
    base = dt.datetime(2026, 2, 16, 0, 0, 0, tzinfo=tz).timestamp()
    hour = 3600.0
    # Vary seconds and sub-second positions so we measure the rounding path too.
    offsets = [0.0, 0.1, 1.0, 29.9, 30.0, 30.1, 59.0, 59.9]
    timestamps: list[float] = []
    for i in range(samples):
        timestamps.append(base + (i * hour / 7.0) + offsets[i % len(offsets)])
    return timestamps


def _measure(
    fn: Callable[[], None],
    rounds: int,
) -> list[float]:
    samples = []
    for _ in range(rounds):
        start = time.perf_counter()
        fn()
        samples.append(time.perf_counter() - start)
    return samples


def _summarize(durations: Sequence[float], operations: int) -> tuple[float, float]:
    median = statistics.median(durations)
    return median, (median / operations) * 1e9


def benchmark_minute_getter(
    timestamps: Sequence[float],
    tz: dt.tzinfo,
    minute_getter: Callable[[float, dt.tzinfo], int],
) -> int:
    acc = 0
    for ts in timestamps:
        acc += minute_getter(ts, tz)
    return acc


def benchmark_hourly_path(
    timestamps: Sequence[float],
    tz: dt.tzinfo,
    minute_getter: Callable[[float, dt.tzinfo], int],
) -> float:
    acc = 0.0
    minute_sets = ([0], [0, 15, 30, 45], [0, 30])
    directions = (True, False)
    for i, ts in enumerate(timestamps):
        acc += _hourly_step(ts, minute_sets[i % len(minute_sets)], tz, directions[i % 2], minute_getter)
    return acc


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--iterations", type=int, default=250_000)
    parser.add_argument("--rounds", type=int, default=7)
    parser.add_argument("--timezones", nargs="*", default=DEFAULT_TIMEZONES)
    args = parser.parse_args()

    print(
        f"Benchmarking with {args.iterations:,} iterations/timezone over {args.rounds} rounds"
    )
    print()

    header = (
        f"{'timezone':<22} {'case':<12} {'old ns/op':>12} "
        f"{'new ns/op':>12} {'slowdown':>10}"
    )
    print(header)
    print("-" * len(header))

    for timezone_name in args.timezones:
        tz = ZoneInfo(timezone_name)
        timestamps = _build_timestamps(tz, args.iterations)

        minute_old = _measure(
            lambda: benchmark_minute_getter(timestamps, tz, _old_minute), args.rounds
        )
        minute_new = _measure(
            lambda: benchmark_minute_getter(timestamps, tz, _new_minute), args.rounds
        )
        minute_old_median, minute_old_ns = _summarize(minute_old, len(timestamps))
        minute_new_median, minute_new_ns = _summarize(minute_new, len(timestamps))

        hourly_old = _measure(
            lambda: benchmark_hourly_path(timestamps, tz, _old_minute), args.rounds
        )
        hourly_new = _measure(
            lambda: benchmark_hourly_path(timestamps, tz, _new_minute), args.rounds
        )
        hourly_old_median, hourly_old_ns = _summarize(hourly_old, len(timestamps))
        hourly_new_median, hourly_new_ns = _summarize(hourly_new, len(timestamps))

        print(
            f"{timezone_name:<22} {'minute':<12} {minute_old_ns:>12.1f} "
            f"{minute_new_ns:>12.1f} {minute_new_median / minute_old_median:>10.2f}x"
        )
        print(
            f"{timezone_name:<22} {'hourly_path':<12} {hourly_old_ns:>12.1f} "
            f"{hourly_new_ns:>12.1f} {hourly_new_median / hourly_old_median:>10.2f}x"
        )


if __name__ == "__main__":
    main()

results.txt

 λ python benchmark.py
Benchmarking with 250,000 iterations/timezone over 7 rounds

timezone               case            old ns/op    new ns/op   slowdown
------------------------------------------------------------------------
UTC                    minute              107.3        174.0       1.62x
UTC                    hourly_path         348.3        498.9       1.43x
Asia/Kolkata           minute              108.9        177.6       1.63x
Asia/Kolkata           hourly_path         336.2        499.5       1.49x
Asia/Kathmandu         minute              111.9        183.2       1.64x
Asia/Kathmandu         hourly_path         349.3        517.1       1.48x
Australia/Adelaide     minute              112.4        201.3       1.79x
Australia/Adelaide     hourly_path         347.7        525.4       1.51x
America/St_Johns       minute              109.8        197.0       1.79x
America/St_Johns       hourly_path         347.4        523.0       1.51x
Pacific/Chatham        minute              111.2        198.6       1.79x
Pacific/Chatham        hourly_path         348.7        521.5       1.50x
Europe/Berlin          minute              111.8        193.5       1.73x
Europe/Berlin          hourly_path         348.5        524.9       1.51x
America/New_York       minute              114.4        204.0       1.78x
America/New_York       hourly_path         354.7        538.0       1.52x