Fixed-Interval Python Tasks with monotonic: Avoid Drift and Skip Missed Slots

A monitoring loop often runs a task and then calls sleep(60). If the task takes 8 seconds, starts are roughly 68 seconds apart, drifting further from the intended cadence. This beginner-to-intermediate tutorial uses Python’s standard library for fixed start intervals, sequential execution, and skipped missed slots.

Use it for short monitoring sessions, local-state polling, and a finite number of samples. No cron changes, system services, or dependencies are needed.

1. Method: wait for scheduled times

With a one-second interval, the first execution starts immediately. A quick task should aim for relative start times of 0, 1, and 2 seconds, rather than waiting a full second after each completion.

Python’s official time documentation states that time.monotonic() cannot go backwards and is unaffected by system-clock updates. Its reference point has no calendar meaning; use differences between readings for durations. Use calendar time for dates, and never subtract a file modification timestamp from a monotonic reading.

The scheduling policy is:

1. Start immediately, then advance scheduled times along the original timeline.
2. Run one task at a time. After completion, skip scheduled times already in the past.
3. An exact scheduled-time match may run immediately; earlier missed tasks are never replayed.
4. --count counts actual calls. A task exception stops the loop with a nonzero exit code.

For a task lasting 2.5 seconds with a one-second interval, starts are roughly 0, 3, and 6 seconds. Slots at 1 and 2 seconds are skipped. A queue where every record must be processed needs persisted pending records rather than this missed-slot policy.

2. Setup: create a demo directory

Use Python 3.9 or newer. These examples were tested on Linux with Python 3.11.4; shell commands work in bash and zsh. Choose a fresh directory name when repeating the demo:

python3 --version
mkdir periodic-demo
cd periodic-demo

Save the next section as periodic.py in this directory. The demo task prints JSON and sleeps to simulate work, without accessing a network or business files.

3. Implementation: fixed intervals and skipped overruns

"""Run a finite, sequential task on a monotonic schedule. Python 3.9+."""
import argparse
import json
import math
import sys
import time


def seconds(value):
    number = float(value)
    if not math.isfinite(number) or not 0 <= number <= 86400:
        raise argparse.ArgumentTypeError("must be finite and between 0 and 86400")
    return number


def positive_int(value):
    number = int(value)
    if number < 1:
        raise argparse.ArgumentTypeError("must be at least 1")
    return number


def run_periodic(action, interval, count, clock=time.monotonic, sleep=time.sleep):
    deadline = clock()
    for number in range(1, count + 1):
        while True:
            remaining = deadline - clock()
            if remaining <= 0:
                break
            sleep(remaining)
        action(number)
        if number == count:
            break
        deadline += interval
        now = clock()
        if deadline < now:
            missed = math.ceil((now - deadline) / interval)
            deadline += missed * interval


def main():
    parser = argparse.ArgumentParser(description=__doc__, allow_abbrev=False)
    parser.add_argument("--interval", type=seconds, default=1.0,
                        help="start interval in seconds, 0.001 to 86400")
    parser.add_argument("--count", type=positive_int, default=3,
                        help="number of actual task executions (default: 3)")
    parser.add_argument("--work-seconds", type=seconds, default=0.25,
                        help="simulated task duration (default: 0.25)")
    parser.add_argument("--fail-at", type=positive_int,
                        help="simulate a failure at this execution number")
    args = parser.parse_args()
    if args.interval < 0.001:
        parser.error("--interval must be at least 0.001 seconds")
    started = time.monotonic()

    def action(number):
        if number == args.fail_at:
            raise RuntimeError("simulated task failure")
        print(json.dumps({"run": number, "elapsed_seconds": round(
            time.monotonic() - started, 3)}), flush=True)
        time.sleep(args.work_seconds)

    try:
        run_periodic(action, args.interval, args.count)
    except KeyboardInterrupt:
        return 130
    except Exception as exc:
        print(f"ERROR: {exc}", file=sys.stderr)
        return 1
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

The key is deadline += interval: advance from the previous schedule, rather than recalculating from task completion. math.ceil() determines how many slots to skip. The while loop recalculates the remaining wait after sleeping, avoiding an unconditional negative sleep() call.

CLI seconds must be finite and within 0–86400; intervals must also be at least 0.001 seconds. This rejects nan, inf, and negative values. The minimum is an input limit, not a precision guarantee. Validation uses math.isfinite() and argparse custom types.

For a real check, replace the printing and simulated sleep inside action(), preserving exceptions on failure. A report check can reuse the rules in checking output age and empty files. Network calls or external commands need their own timeouts: this loop cannot interrupt an action() that has already hung.

4. Usage: observe quick and slow tasks

1. Read the help and run a quick task:

python3 periodic.py --help
python3 periodic.py --interval 1 --work-seconds 0.25 --count 3

Typical output follows; timings vary with machine load. Each line marks a task start, not its completion:

{"run": 1, "elapsed_seconds": 0.0}
{"run": 2, "elapsed_seconds": 1.0}
{"run": 3, "elapsed_seconds": 2.0}

2. Increase simulated work to 2.5 seconds:

python3 periodic.py --interval 1 --work-seconds 2.5 --count 3

Starts should be roughly 0, 3, and 6 seconds. The process exits after the final task completes, so total duration is about 8.5 seconds. There is no extra wait after the last task and no parallel catch-up execution.

The official sleep() documentation notes that scheduling can make a wait longer than requested. If the process wakes late, this example still runs one late task, then skips further expired slots. It is not a real-time scheduler. Restarting creates a new starting point, without persisting progress or guaranteeing execution at a particular calendar time each day.

5. Validation: check boundaries with a fake clock

Save the following as check_periodic.py in the same directory. An injected clock advances a number rather than waiting, immediately checking quick work, overruns, and an exact boundary. Subprocess checks then verify exit codes:

import subprocess
import sys
from periodic import run_periodic


class FakeClock:
    def __init__(self):
        self.now = 0.0

    def clock(self):
        return self.now

    def sleep(self, seconds):
        assert seconds > 0
        self.now += seconds


for duration, expected in [
    (0.25, [0.0, 1.0, 2.0]),
    (2.5, [0.0, 3.0, 6.0]),
    (2.0, [0.0, 2.0, 4.0]),
]:
    fake = FakeClock()
    starts = []

    def action(number):
        starts.append(fake.now)
        fake.now += duration

    run_periodic(action, 1.0, 3, clock=fake.clock, sleep=fake.sleep)
    assert starts == expected, (duration, starts)

base = [sys.executable, "periodic.py"]
for options, expected_code in [
    (["--interval", "0.01", "--work-seconds", "0", "--count", "2"], 0),
    (["--interval", "0"], 2),
    (["--interval", "nan"], 2),
    (["--interval", "inf"], 2),
    (["--count", "0"], 2),
    (["--work-seconds", "-1"], 2),
    (["--work-seconds", "0", "--fail-at", "1"], 1),
]:
    result = subprocess.run(base + options, capture_output=True, text=True,
                            timeout=5)
    assert result.returncode == expected_code, (options, result)
print("PASS: fast, overrun, exact boundary, CLI errors, task failure")

Run it:

python3 check_periodic.py

Expected output:

PASS: fast, overrun, exact boundary, CLI errors, task failure

The fake clock checks scheduling rules; the real run above checks integration and output. Neither guarantees wake-up precision on a production machine.

6. Troubleshooting: separate arguments, work, and scheduling

These commands should exit 2 and 1 respectively. Have the caller inspect exit codes rather than treating missing further output as success:

python3 periodic.py --interval 0
python3 periodic.py --work-seconds 0 --fail-at 1

Normal completion exits 0; Ctrl+C exits 130. This example installs no SIGTERM handler. For finishing a current record before stopping a service, see handling SIGTERM in Python. If an existing scheduling entry cannot find the interpreter or files, use cron PATH and working-directory troubleshooting to fix paths, and avoid stacking an external repeating trigger on this loop.

7. Summary

Schedule relative intervals with monotonic readings and advance from the original timeline to avoid adding each task’s duration to the waiting period. Define how slow tasks skip expired slots, and stop with a nonzero status on exceptions. Verify boundaries with a fake clock, observe real start times, then plug in a check with its own timeouts and error handling.