⏰ Pattern 10: Timers & Periodic Tasks

The Problem

Two flavors: a one-shot delayed call, and a repeating periodic task. The stdlib gives you threading.Timer for the first; the second you build yourself (and the way you build it matters a lot for clean shutdown).

One-Shot: threading.Timer

threading.Timer(interval, function) is a Thread subclass that sleeps for interval seconds and then calls function — exactly once. You .start() it like any thread, and you can .cancel() it before it fires.

import threading

def remind():
    print("⏰ time's up!")

t = threading.Timer(5.0, remind)   # fire once, 5 seconds from now
t.start()                          # returns immediately; countdown runs in a thread

# Change your mind before it fires:
t.cancel()                         # no-op if it has already run

Repeating, Attempt #1: Re-arm Inside the Callback

A Timer fires once. The obvious way to make it repeat is to schedule a fresh Timer at the end of each callback.

import threading

def tick():
    print("tick")
    global timer
    timer = threading.Timer(1.0, tick)   # re-arm for the next second
    timer.start()

timer = threading.Timer(1.0, tick)
timer.start()

# To stop: timer.cancel()  — but you can only cancel the CURRENTLY pending one.

Repeating, The Preferred Way: One Loop Thread + Event.wait()

Run a single thread that loops: do the work, then stop_event.wait(interval). That wait is the delay and the cancel — it sleeps for the interval but returns instantly if someone sets the event. One thread, one clean stop signal. (This is the same responsive shutdown trick from Pattern 9.)

import threading
import time

class RepeatingTimer:
    """Call `function` every `interval` seconds until stopped. Cancellable
    instantly via a threading.Event."""

    def __init__(self, interval: float, function):
        self.interval = interval
        self.function = function
        self._stop = threading.Event()
        self._thread = threading.Thread(target=self._run, daemon=True)

    def _run(self):
        # wait() returns True the moment the event is set, else False on timeout.
        # So the loop ends promptly on stop() instead of after a full interval.
        while not self._stop.wait(self.interval):
            self.function()

    def start(self):
        self._thread.start()

    def stop(self):
        self._stop.set()          # wake the wait() immediately
        self._thread.join()       # and wait for the loop to actually exit


def flush_metrics():
    print(f"flushing metrics at {time.strftime('%X')}")

timer = RepeatingTimer(1.0, flush_metrics)
timer.start()

time.sleep(3.5)                   # let it fire ~3 times
timer.stop()                      # stops within milliseconds, cleanly
print("stopped")

The Periodic Loop, Visualized

The loop lives entirely in wait() between runs. Every iteration either times out (do the work, loop again) or sees the stop event (fall out, thread ends). There's exactly one thread and one decision point.

Drift: Fixed-Delay vs Fixed-Rate

The loop above waits interval seconds after each run finishes. If the work takes 200ms, the real period is interval + 0.2s. Over many iterations this drifts. That's fixed-delay scheduling.

For fixed-rate, compute the next deadline and wait until it — subtracting the time already spent:

import threading
import time

def run_fixed_rate(interval, function, stop_event):
    next_run = time.monotonic()
    while not stop_event.is_set():
        function()
        next_run += interval                       # anchor to the schedule, not to "now"
        delay = next_run - time.monotonic()        # time left until the next tick
        if delay > 0:
            stop_event.wait(delay)                  # cancellable sleep
        # if delay <= 0 we're behind schedule → run again immediately (no sleep)

The Stdlib Alternative: sched

For running many callbacks at different future times from a single thread, the stdlib sched module is a small event scheduler — no thread-per-timer.

import sched
import time

scheduler = sched.scheduler(time.monotonic, time.sleep)
scheduler.enter(2.0, priority=1, action=print, argument=("2s later",))
scheduler.enter(1.0, priority=1, action=print, argument=("1s later",))
scheduler.run()   # blocks, firing each event at its time, in order

Cleaning Up Timers on Shutdown

A pending Timer or a running loop thread is a live thread — a non-daemon one will keep the process from exiting. Always tear timers down.

# One-shot Timer: cancel it if it hasn't fired.
t = threading.Timer(30.0, cleanup)
t.start()
# ... on shutdown:
t.cancel()          # cancels the pending wait; harmless if already fired

# Loop-thread timer: set the event and join.
timer = RepeatingTimer(5.0, poll)
timer.start()
# ... on shutdown:
timer.stop()        # set() + join() → exits within milliseconds

Key Takeaways