+-+-+ +-+ +-+-+-+-+
|G|O| |4| |H|I|V|E|
+-+-+ +-+ +-+-+-+-+

 --- A GOPHER-LIKE INTERFACE FOR HIVE BLOCKCHAIN ---

Learn Zig Series (#70) - Timers and Scheduling

BY: @scipio | CREATED: June 5, 2026, 4:36 p.m. | VOTES: 12 | PAYOUT: $1.08 | [ VOTE ]

Learn Zig Series (#70) - Timers and Scheduling

[IMAGE: https://images.hive.blog/DQmaHuB6qTWHaSpJHQ1S8FCCRmNQuUxTcPZdU4yKHsJ7vEP/zig-banner.png]

What will I learn

Requirements

Difficulty

Curriculum (of the Learn Zig Series):

Learn Zig Series (#70) - Timers and Scheduling

Solutions to Episode 69 Exercises

Exercise 1: Unix socket control interface for the daemon

const std = @import("std");
const posix = std.posix;
const linux = std.os.linux;
const c = @cImport({
    @cInclude("unistd.h");
});

const PID_FILE = "/tmp/zig_service.pid";
const LOG_FILE = "/tmp/zig_service.log";
const HEALTH_FILE = "/tmp/zig_service.health";
const SOCK_PATH = "/tmp/zig_service.sock";

var sig_pipe_fd: posix.fd_t = -1;
var start_time: i64 = 0;
var total_jobs: u64 = 0;

var job_queue: [32]([128]u8) = undefined;
var job_lengths: [32]u16 = [_]u16{0} ** 32;
var job_count: u32 = 0;

fn signalHandler(sig: c_int) callconv(.c) void {
    const byte = [_]u8{@intCast(@as(u32, @bitCast(sig)))};
    _ = posix.write(sig_pipe_fd, &byte) catch {};
}

fn logMsg(log: *std.fs.File, comptime fmt: []const u8, args: anytype) void {
    var buf: [512]u8 = undefined;
    const msg = std.fmt.bufPrint(&buf, fmt, args) catch return;
    var line_buf: [600]u8 = undefined;
    const ts = @divTrunc(std.time.timestamp(), 1);
    const line = std.fmt.bufPrint(&line_buf, "[{d}] {s}\n", .{ ts, msg }) catch return;
    _ = log.write(line) catch {};
}

fn handleClient(cfd: posix.fd_t, log: *std.fs.File, should_stop: *bool) void {
    var buf: [256]u8 = undefined;
    const n = posix.read(cfd, &buf) catch return;
    if (n == 0) return;
    const cmd = std.mem.trim(u8, buf[0..n], " \n\r\t");

    if (std.mem.eql(u8, cmd, "status")) {
        var resp: [256]u8 = undefined;
        const uptime = @divTrunc(std.time.timestamp(), 1) - start_time;
        const r = std.fmt.bufPrint(&resp, "pid={d} uptime={d}s jobs={d} queued={d}\n", .{
            linux.getpid(), uptime, total_jobs, job_count,
        }) catch return;
        _ = posix.write(cfd, r) catch {};
    } else if (std.mem.startsWith(u8, cmd, "queue ")) {
        const text = cmd[6..];
        if (job_count < 32 and text.len < 128) {
            @memcpy(job_queue[job_count][0..text.len], text);
            job_lengths[job_count] = @intCast(text.len);
            job_count += 1;
            _ = posix.write(cfd, "queued\n") catch {};
            logMsg(log, "queued via socket: {s}", .{text});
        } else {
            _ = posix.write(cfd, "queue full\n") catch {};
        }
    } else if (std.mem.eql(u8, cmd, "stop")) {
        _ = posix.write(cfd, "stopping\n") catch {};
        logMsg(log, "stop command received via socket", .{});
        should_stop.* = true;
    } else {
        _ = posix.write(cfd, "unknown command\n") catch {};
    }
}

pub fn main() !void {
    const stdout = std.io.getStdOut().writer();
    start_time = @divTrunc(std.time.timestamp(), 1);

    var log = try std.fs.cwd().createFile(LOG_FILE, .{ .truncate = false });
    defer log.close();
    log.seekFromEnd(0) catch {};

    // PID file
    { var f = try std.fs.cwd().createFile(PID_FILE, .{}); defer f.close();
      var pb: [32]u8 = undefined;
      const ps = std.fmt.bufPrint(&pb, "{d}\n", .{linux.getpid()}) catch unreachable;
      try f.writeAll(ps); }
    defer std.fs.cwd().deleteFile(PID_FILE) catch {};

    // signal pipe
    const pipe_fds = try posix.pipe();
    sig_pipe_fd = pipe_fds[1];
    var flags = linux.fcntl(pipe_fds[1], linux.F.GETFL, @as(linux.fd_t, 0));
    _ = linux.fcntl(pipe_fds[1], linux.F.SETFL, flags | @as(u32, @bitCast(linux.O{ .NONBLOCK = true })));
    var sa: linux.Sigaction = .{
        .handler = .{ .handler = signalHandler },
        .mask = linux.empty_sigset,
        .flags = linux.SA.RESTART,
    };
    _ = linux.sigaction(linux.SIG.TERM, &sa, null);

    // control socket
    std.fs.cwd().deleteFile(SOCK_PATH) catch {};
    const sock_fd = try posix.socket(posix.AF.UNIX, posix.SOCK.STREAM, 0);
    defer posix.close(sock_fd);
    const addr = try std.net.Address.initUnix(SOCK_PATH);
    try posix.bind(sock_fd, &addr.any, addr.getOsSockLen());
    try posix.listen(sock_fd, 5);
    defer std.fs.cwd().deleteFile(SOCK_PATH) catch {};

    try stdout.print("Daemon pid {d}, socket {s}\n", .{ linux.getpid(), SOCK_PATH });
    logMsg(&log, "daemon started pid={d}", .{linux.getpid()});

    var should_stop = false;
    while (!should_stop) {
        var pollfds = [2]linux.pollfd{
            .{ .fd = pipe_fds[0], .events = linux.POLL.IN, .revents = 0 },
            .{ .fd = sock_fd, .events = linux.POLL.IN, .revents = 0 },
        };
        _ = linux.poll(&pollfds, 2, 2000);

        if (pollfds[0].revents & linux.POLL.IN != 0) {
            var sb: [16]u8 = undefined;
            _ = posix.read(pipe_fds[0], &sb) catch {};
            should_stop = true;
        }

        if (pollfds[1].revents & linux.POLL.IN != 0) {
            var ca: posix.sockaddr = undefined;
            var cl: posix.socklen_t = @sizeOf(posix.sockaddr);
            if (posix.accept(sock_fd, &ca, &cl)) |cfd| {
                defer posix.close(cfd);
                handleClient(cfd, &log, &should_stop);
            } else |_| {}
        }

        // process in-memory queue
        while (job_count > 0) {
            job_count -= 1;
            const len = job_lengths[job_count];
            logMsg(&log, "processed: {s}", .{job_queue[job_count][0..len]});
            total_jobs += 1;
        }
    }

    logMsg(&log, "daemon stopped, total jobs={d}", .{total_jobs});
    posix.close(pipe_fds[0]);
    posix.close(pipe_fds[1]);
}

The daemon multiplexes the signal pipe and control socket in a single poll() call. Client commands are parsed as simple text -- status returns runtime stats, queue pushes a job into a fixed-size in-memory ring, and stop sets the termination flag. The in-memory queue is drained each tick alongside the file-based queue from the original.

Exercise 2: Log rotation with size threshold and SIGUSR1

const std = @import("std");
const posix = std.posix;
const linux = std.os.linux;

const LOG_PATH = "/tmp/zig_service.log";
const LOG_ROTATED = "/tmp/zig_service.log.1";
const MAX_LOG_BYTES: u64 = 10_000;

var sig_pipe_fd: posix.fd_t = -1;
var bytes_written: u64 = 0;

fn signalHandler(sig: c_int) callconv(.c) void {
    const byte = [_]u8{@intCast(@as(u32, @bitCast(sig)))};
    _ = posix.write(sig_pipe_fd, &byte) catch {};
}

fn rotateLog(log: *std.fs.File) void {
    log.close();
    std.fs.cwd().deleteFile(LOG_ROTATED) catch {};
    std.fs.cwd().rename(LOG_PATH, LOG_ROTATED) catch {};
    log.* = std.fs.cwd().createFile(LOG_PATH, .{}) catch return;
    bytes_written = 0;
}

fn writeLog(log: *std.fs.File, comptime fmt: []const u8, args: anytype) void {
    var buf: [512]u8 = undefined;
    const msg = std.fmt.bufPrint(&buf, fmt, args) catch return;
    var line_buf: [600]u8 = undefined;
    const line = std.fmt.bufPrint(&line_buf, "[{d}] {s}\n", .{
        @divTrunc(std.time.timestamp(), 1), msg,
    }) catch return;
    const n = log.write(line) catch return;
    bytes_written += n;

    if (bytes_written >= MAX_LOG_BYTES) {
        rotateLog(log);
        writeLog(log, "log rotated (size threshold)", .{});
    }
}

pub fn main() !void {
    const stdout = std.io.getStdOut().writer();

    const pipe_fds = try posix.pipe();
    sig_pipe_fd = pipe_fds[1];
    var flags = linux.fcntl(pipe_fds[1], linux.F.GETFL, @as(linux.fd_t, 0));
    _ = linux.fcntl(pipe_fds[1], linux.F.SETFL, flags | @as(u32, @bitCast(linux.O{ .NONBLOCK = true })));

    var sa: linux.Sigaction = .{
        .handler = .{ .handler = signalHandler },
        .mask = linux.empty_sigset,
        .flags = linux.SA.RESTART,
    };
    _ = linux.sigaction(linux.SIG.USR1, &sa, null);
    _ = linux.sigaction(linux.SIG.TERM, &sa, null);

    var log = try std.fs.cwd().createFile(LOG_PATH, .{ .truncate = false });
    defer log.close();
    log.seekFromEnd(0) catch {};
    bytes_written = log.getPos() catch 0;

    try stdout.print("Log rotation demo (pid {d}). SIGUSR1 = force rotate, SIGTERM = stop.\n", .{linux.getpid()});
    var should_stop = false;
    var tick: u32 = 0;

    while (!should_stop) {
        var pollfds = [_]linux.pollfd{.{ .fd = pipe_fds[0], .events = linux.POLL.IN, .revents = 0 }};
        _ = linux.poll(&pollfds, 1, 500);

        if (pollfds[0].revents & linux.POLL.IN != 0) {
            var buf: [16]u8 = undefined;
            const n = posix.read(pipe_fds[0], &buf) catch 0;
            for (buf[0..n]) |sig| {
                if (sig == @as(u8, @intCast(linux.SIG.USR1))) {
                    rotateLog(&log);
                    writeLog(&log, "log rotated (SIGUSR1)", .{});
                } else if (sig == @as(u8, @intCast(linux.SIG.TERM))) {
                    should_stop = true;
                }
            }
        }

        tick += 1;
        writeLog(&log, "tick {d}, bytes_written={d}", .{ tick, bytes_written });
    }

    writeLog(&log, "stopped after {d} ticks", .{tick});
    posix.close(pipe_fds[0]);
    posix.close(pipe_fds[1]);
}

The rotateLog function closes the current file, renames it to .1, and opens a fresh log. writeLog tracks cumulative bytes and triggers rotation when the threshold is crossed. SIGUSR1 forces immediate rotation via the self-pipe. The initial byte count is seeded from the existing file position so we don't lose track after a restart.

Exercise 3: Daemon supervisor with health file monitoring

const std = @import("std");
const posix = std.posix;
const linux = std.os.linux;

const HEALTH_FILE = "/tmp/zig_service.health";
const PID_FILE = "/tmp/zig_service.pid";
const DAEMON_BIN = "/tmp/zig_service";
const SUP_LOG = "/tmp/zig_supervisor.log";

fn supLog(log: *std.fs.File, comptime fmt: []const u8, args: anytype) void {
    var buf: [512]u8 = undefined;
    const msg = std.fmt.bufPrint(&buf, fmt, args) catch return;
    var line_buf: [600]u8 = undefined;
    const line = std.fmt.bufPrint(&line_buf, "[{d}] {s}\n", .{
        @divTrunc(std.time.timestamp(), 1), msg,
    }) catch return;
    _ = log.write(line) catch {};
}

fn readHealthTimestamp() ?i64 {
    var file = std.fs.cwd().openFile(HEALTH_FILE, .{}) catch return null;
    defer file.close();
    var buf: [32]u8 = undefined;
    const n = file.read(&buf) catch return null;
    const trimmed = std.mem.trim(u8, buf[0..n], " \n\r\t");
    return std.fmt.parseInt(i64, trimmed, 10) catch null;
}

fn readPid() ?i32 {
    var file = std.fs.cwd().openFile(PID_FILE, .{}) catch return null;
    defer file.close();
    var buf: [32]u8 = undefined;
    const n = file.read(&buf) catch return null;
    const trimmed = std.mem.trim(u8, buf[0..n], " \n\r\t");
    return std.fmt.parseInt(i32, trimmed, 10) catch null;
}

pub fn main() !void {
    const stdout = std.io.getStdOut().writer();
    var log = try std.fs.cwd().createFile(SUP_LOG, .{ .truncate = false });
    defer log.close();
    log.seekFromEnd(0) catch {};

    try stdout.print("Supervisor started. Monitoring {s}\n", .{HEALTH_FILE});
    supLog(&log, "supervisor started", .{});

    var consecutive_failures: u32 = 0;

    for (0..60) |_| {
        std.time.sleep(5 * std.time.ns_per_s);
        const now = @divTrunc(std.time.timestamp(), 1);

        if (readHealthTimestamp()) |ts| {
            const age = now - ts;
            if (age > 15) {
                consecutive_failures += 1;
                supLog(&log, "health stale: {d}s old (failure {d}/3)", .{ age, consecutive_failures });

                if (consecutive_failures >= 3) {
                    supLog(&log, "3 consecutive failures, restarting daemon", .{});
                    if (readPid()) |pid| {
                        _ = linux.kill(pid, linux.SIG.TERM);
                        std.time.sleep(2 * std.time.ns_per_s);
                        // check if it actually died
                        const result = linux.kill(pid, 0);
                        const signed: isize = @bitCast(@as(usize, result));
                        if (signed == 0) {
                            _ = linux.kill(pid, linux.SIG.KILL);
                            std.time.sleep(500 * std.time.ns_per_ms);
                        }
                    }
                    supLog(&log, "starting new daemon instance", .{});
                    // in production you'd exec the daemon binary here
                    try stdout.print("[supervisor] would exec {s}\n", .{DAEMON_BIN});
                    consecutive_failures = 0;
                }
            } else {
                if (consecutive_failures > 0) {
                    supLog(&log, "health recovered (was at {d} failures)", .{consecutive_failures});
                }
                consecutive_failures = 0;
            }
        } else {
            consecutive_failures += 1;
            supLog(&log, "no health file found (failure {d}/3)", .{consecutive_failures});
        }
    }

    supLog(&log, "supervisor exiting after 60 checks", .{});
}

The supervisor reads the health timestamp every 5 seconds and compares it to now. If the timestamp is stale (>15s old), the failure counter increments. At 3 consecutive failures, the supervisor reads the PID file, sends SIGTERM, waits for exit, and would exec the daemon binary to restart it. The counter resets on either a successful health check or after a restart.

Last episode we built a full daemon skeleton -- double-fork, PID files, logging, signal handling, health checking. The daemon runs in the background, does work, responds to signals. But there's one big question we barely touched: how does the daemon know WHEN to do things?

Our episode 69 daemon used a hardcoded poll() timeout as its work interval. That works for simple "check every N seconds" loops but falls apart the moment you need precision. What if you need to fire exactly every 100 milliseconds for a heartbeat? What if you need tasks to run at "every Monday at 3am"? What if your system's clock jumps forward by 30 seconds because NTP corrected a drift?

Time is surprisingly tricky in systems programming. There are two completely different clocks on your machine, they disagree with each other, and both of them lie sometimes. Let's sort this out ;-)

Monotonic vs wall clock time

Your computer has (at least) two time sources, and understanding when to use which one is crucial for anything involving scheduling or elapsed time measurement.

Wall clock time (also called "real time" or CLOCK_REALTIME) is what you see when you check the clock on your desktop. It's measured as seconds since January 1, 1970 (the Unix epoch). It can jump forwards when NTP synchronizes your clock, jump backwards if someone sets the clock manually, or leap by a full second during a UTC leap second insertion. Wall clock time is for timestamps that need to correspond to "real world" dates and times -- log entries, scheduling cron jobs, displaying "last modified" dates.

Monotonic time (CLOCK_MONOTONIC) is a counter that only ever goes forward. It typically starts at some arbitrary point (usually system boot) and ticks steadily upward. NTP can't make it jump. Manual clock changes don't affect it. It exists purely for measuring elapsed time between two points. If you start a timer and read it 5 seconds later, monotonic time guarantees you get ~5 seconds. Wall clock time might say 5 seconds, or 35 seconds (NTP correction), or -25 seconds (clock set backwards).

The rule is simple: use monotonic time for durations and intervals, wall clock time for calendar scheduling.

const std = @import("std");
const posix = std.posix;
const linux = std.os.linux;

pub fn main() !void {
    const stdout = std.io.getStdOut().writer();

    // Monotonic time via std.time.Instant
    // this is Zig's cross-platform monotonic clock
    const start = try std.time.Instant.now();

    // simulate some work
    var sum: u64 = 0;
    for (0..10_000_000) |i| {
        sum +%= i;
    }

    const end = try std.time.Instant.now();
    const elapsed_ns = end.since(start);
    const elapsed_ms = @as(f64, @floatFromInt(elapsed_ns)) / 1_000_000.0;
    try stdout.print("Monotonic elapsed: {d:.3}ms (sum={d})\n", .{ elapsed_ms, sum });

    // Wall clock time via std.time.timestamp()
    // returns seconds since epoch (1970-01-01)
    const wall_secs = std.time.timestamp();
    try stdout.print("Wall clock: {d} seconds since epoch\n", .{wall_secs});

    // Nanosecond wall clock via clock_gettime
    var ts: linux.timespec = undefined;
    _ = linux.clock_gettime(linux.CLOCK.REALTIME, &ts);
    try stdout.print("Wall clock (precise): {d}.{d:0>9} seconds\n", .{ ts.sec, ts.nsec });

    // Monotonic raw via clock_gettime (not adjusted by NTP slewing)
    var mono_ts: linux.timespec = undefined;
    _ = linux.clock_gettime(linux.CLOCK.MONOTONIC, &mono_ts);
    try stdout.print("Monotonic: {d}.{d:0>9} seconds since boot\n", .{ mono_ts.sec, mono_ts.nsec });

    // the key difference: monotonic never jumps
    try stdout.print("\nKey insight: Instant.now() uses CLOCK_MONOTONIC internally.\n", .{});
    try stdout.print("Use Instant for measuring HOW LONG something takes.\n", .{});
    try stdout.print("Use timestamp()/CLOCK_REALTIME for WHEN something happened.\n", .{});
}

std.time.Instant is Zig's portable monotonic clock. It wraps CLOCK_MONOTONIC on Linux, mach_absolute_time on macOS, and QueryPerformanceCounter on Windows. The .since() method returns nanoseconds between two instants. For wall clock time, std.time.timestamp() gives you seconds since epoch, or you can use clock_gettime(CLOCK_REALTIME) directly for nanosecond precision.

Having said that, there's a subtlety here. CLOCK_MONOTONIC on Linux is still subject to NTP slewing -- NTP gradually adjusts the tick rate to bring the clock in line, so monotonic time might run very slightly faster or slower than real seconds. If you need a clock that is completely immune to NTP (even slewing), Linux provides CLOCK_MONOTONIC_RAW. For 99.9% of use cases CLOCK_MONOTONIC is fine -- the slew rate is tiny (max 500ppm, or 0.05%).

Measuring elapsed time with std.time.Timer

Zig's standard library includes std.time.Timer -- a convenient wrapper around the monotonic clock that tracks elapsed time with a start/stop/lap interface:

const std = @import("std");

pub fn main() !void {
    const stdout = std.io.getStdOut().writer();

    // Timer auto-starts on creation
    var timer = try std.time.Timer.start();

    // do some work in phases
    var sum: u64 = 0;
    for (0..5_000_000) |i| {
        sum +%= i;
    }

    // lap() returns elapsed since last lap (or start)
    const phase1_ns = timer.lap();
    const phase1_ms = @as(f64, @floatFromInt(phase1_ns)) / 1_000_000.0;
    try stdout.print("Phase 1: {d:.3}ms (sum={d})\n", .{ phase1_ms, sum });

    // do more work
    for (0..10_000_000) |i| {
        sum +%= i;
    }

    const phase2_ns = timer.lap();
    const phase2_ms = @as(f64, @floatFromInt(phase2_ns)) / 1_000_000.0;
    try stdout.print("Phase 2: {d:.3}ms\n", .{phase2_ms});

    // read() returns total elapsed since start (without resetting)
    const total_ns = timer.read();
    const total_ms = @as(f64, @floatFromInt(total_ns)) / 1_000_000.0;
    try stdout.print("Total:   {d:.3}ms\n", .{total_ms});

    // reset() starts the timer over
    timer.reset();
    std.time.sleep(50 * std.time.ns_per_ms);
    const after_reset = timer.read();
    try stdout.print("After reset + 50ms sleep: {d:.3}ms\n", .{
        @as(f64, @floatFromInt(after_reset)) / 1_000_000.0,
    });
}

The Timer is perfect for benchmarking sections of code, profiling hot paths, or implementing timeouts. lap() returns the time since the previous lap and resets the lap counter, making it ideal for measuring sequential phases of work. read() gives total elapsed without resetting anything.

Periodic timers with timerfd

For a daemon that needs to fire callbacks at precise intervals, polling with a timeout works but has a problem: the actual interval is timeout + processing time. If your work takes 50ms and your timeout is 1000ms, you fire every 1050ms. Over hours that drift accumulates.

Linux provides timerfd -- a file descriptor that becomes readable when a timer expires. You create it, set an interval, and then poll() or read() on it. The kernel manages the timing, compensating for processing delay automatically:

const std = @import("std");
const posix = std.posix;
const linux = std.os.linux;

pub fn main() !void {
    const stdout = std.io.getStdOut().writer();

    // create a timerfd using monotonic clock
    const tfd_result = linux.timerfd_create(linux.CLOCK.MONOTONIC, .{});
    const signed: isize = @bitCast(@as(usize, tfd_result));
    if (signed < 0) {
        try stdout.print("timerfd_create failed\n", .{});
        return;
    }
    const tfd: posix.fd_t = @intCast(tfd_result);
    defer posix.close(tfd);

    // set it to fire every 500ms, starting after 500ms
    const spec = linux.itimerspec{
        .it_interval = .{ .sec = 0, .nsec = 500_000_000 }, // period: 500ms
        .it_value = .{ .sec = 0, .nsec = 500_000_000 }, // initial: 500ms
    };
    const set_result = linux.timerfd_settime(@intCast(tfd), .{}, &spec, null);
    const set_signed: isize = @bitCast(@as(usize, set_result));
    if (set_signed < 0) {
        try stdout.print("timerfd_settime failed\n", .{});
        return;
    }

    try stdout.print("Periodic timer: 500ms interval. Running 10 ticks...\n", .{});
    const start = try std.time.Instant.now();

    for (0..10) |i| {
        // read() blocks until the timer fires
        // returns a u64 count of expirations since last read
        var expirations: u64 = 0;
        const bytes = posix.read(tfd, std.mem.asBytes(&expirations)) catch 0;
        _ = bytes;

        const elapsed_ns = (try std.time.Instant.now()).since(start);
        const elapsed_ms = @as(f64, @floatFromInt(elapsed_ns)) / 1_000_000.0;
        try stdout.print("  tick {d}: {d:.1}ms (expirations: {d})\n", .{
            i, elapsed_ms, expirations,
        });
    }

    try stdout.print("Done. Expected ~5000ms, got {d:.1}ms\n", .{
        @as(f64, @floatFromInt((try std.time.Instant.now()).since(start))) / 1_000_000.0,
    });
}

The beauty of timerfd is that it integrates with poll()/epoll() -- you can multiplex timer events with socket events, signal events (via signalfd), and file events (via inotify) all in a single event loop. This is exactly how production event-driven servers work. The read() on the timerfd returns the number of expirations since the last read, so if your processing is slow and misses a tick, the count tells you how many you missed.

NB: timerfd is Linux-specific. On macOS you'd use kqueue with EVFILT_TIMER, and on Windows you'd use CreateWaitableTimer. Zig's standard library doesn't abstract over these yet, so cross-platform code needs conditional compilation or a manual abstraction layer.

Building a simple interval scheduler

Let's build something more useful: a scheduler that runs multiple tasks at different intervals. Each task has a name, a function pointer, and an interval. The scheduler keeps track of when each task last ran and fires them when they're due:

const std = @import("std");

const Task = struct {
    name: []const u8,
    interval_ms: u64,
    last_run_ns: u64,
    run_count: u64,
    callback: *const fn ([]const u8) void,
};

fn taskHeartbeat(name: []const u8) void {
    std.debug.print("[{s}] heartbeat at {d}s\n", .{
        name,
        @divTrunc(std.time.timestamp(), 1),
    });
}

fn taskCleanup(name: []const u8) void {
    std.debug.print("[{s}] cleanup cycle\n", .{name});
}

fn taskReport(name: []const u8) void {
    std.debug.print("[{s}] generating report\n", .{name});
}

pub fn main() !void {
    const stdout = std.io.getStdOut().writer();

    var tasks = [_]Task{
        .{ .name = "heartbeat", .interval_ms = 500, .last_run_ns = 0, .run_count = 0, .callback = taskHeartbeat },
        .{ .name = "cleanup", .interval_ms = 2000, .last_run_ns = 0, .run_count = 0, .callback = taskCleanup },
        .{ .name = "report", .interval_ms = 5000, .last_run_ns = 0, .run_count = 0, .callback = taskReport },
    };

    try stdout.print("Scheduler running 3 tasks for 10 seconds...\n", .{});
    const base = try std.time.Instant.now();

    while (true) {
        const now = try std.time.Instant.now();
        const elapsed_ns = now.since(base);
        const elapsed_ms = elapsed_ns / std.time.ns_per_ms;

        // stop after 10 seconds
        if (elapsed_ms >= 10_000) break;

        // find the soonest task deadline
        var min_sleep_ms: u64 = 100; // default poll interval
        for (&tasks) |*task| {
            const since_last = elapsed_ns - task.last_run_ns;
            const interval_ns = task.interval_ms * std.time.ns_per_ms;

            if (since_last >= interval_ns) {
                // task is due -- run it
                task.callback(task.name);
                task.last_run_ns = elapsed_ns;
                task.run_count += 1;
            } else {
                // how long until this task is due?
                const remaining_ns = interval_ns - since_last;
                const remaining_ms = remaining_ns / std.time.ns_per_ms;
                if (remaining_ms < min_sleep_ms) {
                    min_sleep_ms = remaining_ms;
                }
            }
        }

        // sleep until the next task is due (or 1ms minimum to avoid busy-spin)
        if (min_sleep_ms < 1) min_sleep_ms = 1;
        std.time.sleep(min_sleep_ms * std.time.ns_per_ms);
    }

    try stdout.print("\nFinal counts:\n", .{});
    for (tasks) |task| {
        try stdout.print("  {s}: ran {d} times\n", .{ task.name, task.run_count });
    }
}

The scheduler calculates the sleep duration dynamically -- it sleeps only until the next task is due, not for a fixed interval. This means the heartbeat fires roughly every 500ms even though cleanup and report have longer intervals. After 10 seconds you should see approximately 20 heartbeats, 5 cleanups, and 2 reports.

This is the foundation of every task scheduler. Production versions add priority queues (so you don't scan all tasks every tick), persistent state (so tasks survive restarts), and error handling (so one failed task doesn't block others). But the core loop is always the same: check what's due, run it, sleep until the next deadline.

Parsing cron expressions

For calendar-based scheduling ("every Monday at 3am", "every 15 minutes"), the cron expression is the universal standard. A cron expression has 5 fields: minute, hour, day-of-month, month, day-of-week. Each field can be a number, * (any), or a range/step.

Let's build a simple cron parser that handles numbers, wildcards, and step values:

const std = @import("std");

const CronField = struct {
    // bit set: bit N means "value N is valid"
    // minutes: 0-59, hours: 0-23, mday: 1-31, month: 1-12, wday: 0-6
    bits: u64,

    fn matchesValue(self: CronField, val: u6) bool {
        return (self.bits >> val) & 1 == 1;
    }
};

const CronExpr = struct {
    minute: CronField,
    hour: CronField,
    mday: CronField,
    month: CronField,
    wday: CronField,

    fn matchesNow(self: CronExpr) bool {
        // get current local time from epoch seconds
        const epoch = std.time.timestamp();
        const es = @as(u64, @intCast(epoch));
        // break epoch into components (UTC)
        const secs_in_day: u64 = 86400;
        const days = es / secs_in_day;
        const day_secs = es % secs_in_day;
        const hour: u6 = @intCast(day_secs / 3600);
        const minute: u6 = @intCast((day_secs % 3600) / 60);
        // day of week: Jan 1 1970 was Thursday (4)
        const wday: u6 = @intCast((days + 4) % 7);
        // rough month/mday from days since epoch (good enough for demo)
        const year_days = days % 365; // ignoring leap years for brevity
        const mday: u6 = @intCast((year_days % 30) + 1);
        const month: u6 = @intCast((year_days / 30) + 1);

        return self.minute.matchesValue(minute) and
            self.hour.matchesValue(hour) and
            self.mday.matchesValue(mday) and
            self.month.matchesValue(month) and
            self.wday.matchesValue(wday);
    }
};

fn parseField(field: []const u8, min: u6, max: u6) !CronField {
    var bits: u64 = 0;

    if (std.mem.eql(u8, field, "*")) {
        // wildcard: set all bits in range
        var v = min;
        while (v <= max) : (v += 1) {
            bits |= @as(u64, 1) << v;
            if (v == max) break;
        }
        return CronField{ .bits = bits };
    }

    // check for step: */N
    if (std.mem.startsWith(u8, field, "*/")) {
        const step = std.fmt.parseInt(u6, field[2..], 10) catch return error.InvalidStep;
        if (step == 0) return error.InvalidStep;
        var v = min;
        while (v <= max) {
            bits |= @as(u64, 1) << v;
            const next = @as(u7, v) + step;
            if (next > max) break;
            v = @intCast(next);
        }
        return CronField{ .bits = bits };
    }

    // single value
    const val = std.fmt.parseInt(u6, field, 10) catch return error.InvalidValue;
    if (val < min or val > max) return error.OutOfRange;
    bits |= @as(u64, 1) << val;
    return CronField{ .bits = bits };
}

fn parseCron(expr: []const u8) !CronExpr {
    var fields: [5][]const u8 = undefined;
    var count: usize = 0;
    var iter = std.mem.splitScalar(u8, expr, ' ');
    while (iter.next()) |part| {
        if (part.len == 0) continue;
        if (count >= 5) return error.TooManyFields;
        fields[count] = part;
        count += 1;
    }
    if (count != 5) return error.WrongFieldCount;

    return CronExpr{
        .minute = try parseField(fields[0], 0, 59),
        .hour = try parseField(fields[1], 0, 23),
        .mday = try parseField(fields[2], 1, 31),
        .month = try parseField(fields[3], 1, 12),
        .wday = try parseField(fields[4], 0, 6),
    };
}

pub fn main() !void {
    const stdout = std.io.getStdOut().writer();

    const exprs = [_][]const u8{
        "*/15 * * * *",  // every 15 minutes
        "0 3 * * 1",     // Monday at 3:00
        "30 */2 * * *",  // :30 past every 2nd hour
    };

    for (exprs) |expr| {
        const cron = parseCron(expr) catch {
            try stdout.print("Failed to parse: {s}\n", .{expr});
            continue;
        };
        const matches = cron.matchesNow();
        try stdout.print("'{s}' matches now: {}\n", .{ expr, matches });
    }

    // demonstrate bit matching
    const every15 = try parseCron("*/15 * * * *");
    try stdout.print("\n*/15 minute bits: ", .{});
    for (0..60) |m| {
        if (every15.minute.matchesValue(@intCast(m))) {
            try stdout.print("{d} ", .{m});
        }
    }
    try stdout.print("\n", .{});
}

The bit-set representation is elegant -- each possible value gets one bit in a u64, and checking if a time component matches is just a single bit test. Parsing */15 for the minute field sets bits 0, 15, 30, 45. Parsing * sets all bits in the valid range. A single number sets one bit.

The matchesNow function is simplified (it ignores leap years and does rough month calculation), but the matching logic is correct. A production cron engine would use proper calendar math from a library, but the core idea -- parse to bit sets, AND them against the current time components -- is exactly what real cron implementations do.

Timer wheels: efficient timer management at scale

What if you have thousands of active timers? Scanning all of them every tick is O(n), which gets expensive fast. A timer wheel gives you O(1) insertion and O(1) expiration by organizing timers into slots like a clock face.

The idea: imagine a wheel with 256 slots, where each slot represents one tick of your base resolution. Slot 0 is "now", slot 1 is "1 tick from now", etc. To insert a timer that fires in 50 ticks, you put it in slot (current + 50) % 256. Every tick, you advance the wheel pointer by one and fire everything in that slot.

const std = @import("std");

const WHEEL_SIZE = 256;

const TimerEntry = struct {
    id: u32,
    name: [32]u8,
    name_len: u8,
    remaining_rounds: u32, // for timers > WHEEL_SIZE ticks away
    callback_id: u8,       // which callback to invoke
};

const TimerWheel = struct {
    slots: [WHEEL_SIZE][8]TimerEntry,
    slot_counts: [WHEEL_SIZE]u8,
    current: u32,
    total_fired: u64,

    fn init() TimerWheel {
        var wheel = TimerWheel{
            .slots = undefined,
            .slot_counts = [_]u8{0} ** WHEEL_SIZE,
            .current = 0,
            .total_fired = 0,
        };
        return wheel;
    }

    fn insert(self: *TimerWheel, id: u32, name: []const u8, ticks: u32, cb_id: u8) bool {
        const slot_idx = (self.current + (ticks % WHEEL_SIZE)) % WHEEL_SIZE;
        const rounds = ticks / WHEEL_SIZE;
        const count = self.slot_counts[slot_idx];
        if (count >= 8) return false; // slot full

        var entry = &self.slots[slot_idx][count];
        entry.id = id;
        entry.remaining_rounds = rounds;
        entry.callback_id = cb_id;
        entry.name_len = @intCast(@min(name.len, 32));
        @memcpy(entry.name[0..entry.name_len], name[0..entry.name_len]);
        self.slot_counts[slot_idx] += 1;
        return true;
    }

    fn tick(self: *TimerWheel) void {
        const slot = self.current;
        var i: u8 = 0;
        var kept: u8 = 0;

        while (i < self.slot_counts[slot]) {
            var entry = &self.slots[slot][i];
            if (entry.remaining_rounds == 0) {
                // timer expired -- fire it
                std.debug.print("  FIRE: id={d} name={s}\n", .{
                    entry.id,
                    entry.name[0..entry.name_len],
                });
                self.total_fired += 1;
                i += 1;
            } else {
                // not yet -- decrement and keep
                entry.remaining_rounds -= 1;
                if (kept != i) {
                    self.slots[slot][kept] = self.slots[slot][i];
                }
                kept += 1;
                i += 1;
            }
        }
        self.slot_counts[slot] = kept;
        self.current = (self.current + 1) % WHEEL_SIZE;
    }
};

pub fn main() !void {
    const stdout = std.io.getStdOut().writer();

    var wheel = TimerWheel.init();

    // add timers at various delays
    _ = wheel.insert(1, "fast", 3, 0);          // fires in 3 ticks
    _ = wheel.insert(2, "medium", 10, 0);       // fires in 10 ticks
    _ = wheel.insert(3, "slow", 260, 0);        // fires in 260 ticks (>1 revolution)
    _ = wheel.insert(4, "very-fast", 1, 0);     // fires in 1 tick
    _ = wheel.insert(5, "exact-wheel", 256, 0); // fires in exactly 256 ticks

    try stdout.print("Timer wheel demo ({d} slots). Running 270 ticks...\n\n", .{WHEEL_SIZE});

    for (0..270) |t| {
        if (t < 15 or t > 255) {
            try stdout.print("tick {d}:\n", .{t});
        }
        wheel.tick();
    }

    try stdout.print("\nTotal timers fired: {d}\n", .{wheel.total_fired});
}

The remaining_rounds field handles timers that are further out than WHEEL_SIZE ticks. When the wheel pointer reaches a slot, it checks each entry: if rounds is 0, fire it. If rounds > 0, decrement and keep it for the next revolution. This gives O(1) amortized insertion and expiration even for millions of timers.

Timer wheels are used in the Linux kernel (for network timeouts), in Netty (the Java NIO framework), in Kafka (for delayed message delivery), and in basically every high-performance server that manages many concurrent timeouts. The implementation above is simplified (fixed 8 entries per slot) but demonstrates the core algorithim. A production version would use linked lists per slot and support hierarchical wheels for very long timers.

Handling clock jumps and NTP adjustments

Here's a scenario that bites people: your daemon uses wall clock time to schedule a task at 3:00 AM. At 2:58 AM, NTP realizes the system clock is 5 minutes ahead and jumps it back to 2:53 AM. Your daemon thinks 3:00 AM hasn't happened yet and waits... but it already fired the task at the "first" 2:58 AM. Result: the task runs twice.

Or the opposite: the clock jumps forward by 10 minutes, skipping 3:00 AM entirely. The task never fires.

The solution is to use monotonic time for the sleep/wait intervals and only consult wall clock time when deciding WHAT to run:

const std = @import("std");
const linux = std.os.linux;

const ClockMonitor = struct {
    last_wall: i64,
    last_mono_ns: u64,
    jump_threshold_s: i64,
    jump_count: u32,

    fn init(threshold_s: i64) !ClockMonitor {
        const wall = std.time.timestamp();
        const mono = (try std.time.Instant.now()).since(
            try std.time.Instant.now()
        );
        // just use 0 as base for mono
        return ClockMonitor{
            .last_wall = wall,
            .last_mono_ns = 0,
            .jump_threshold_s = threshold_s,
            .jump_count = 0,
        };
    }

    fn check(self: *ClockMonitor) !?struct { direction: []const u8, delta_s: i64 } {
        const now_wall = std.time.timestamp();
        const wall_delta = now_wall - self.last_wall;

        // we expect wall delta to be roughly equal to our poll interval
        // if it's way off, the clock jumped
        if (wall_delta < 0) {
            // clock went backwards!
            self.jump_count += 1;
            const result = .{ .direction = "BACKWARD", .delta_s = wall_delta };
            self.last_wall = now_wall;
            return result;
        }

        if (wall_delta > self.jump_threshold_s) {
            // clock jumped forward suspiciously far
            self.jump_count += 1;
            const result = .{ .direction = "FORWARD", .delta_s = wall_delta };
            self.last_wall = now_wall;
            return result;
        }

        self.last_wall = now_wall;
        return null;
    }
};

pub fn main() !void {
    const stdout = std.io.getStdOut().writer();

    var monitor = try ClockMonitor.init(5); // alert on jumps > 5 seconds

    try stdout.print("Clock monitor running. Checking every second for 15 iterations.\n", .{});
    try stdout.print("(To test: run 'sudo date -s +10min' in another terminal)\n\n", .{});

    for (0..15) |i| {
        const wall = std.time.timestamp();

        if (try monitor.check()) |jump| {
            try stdout.print("[{d}] CLOCK JUMP DETECTED: {s} by {d}s! (jump #{d})\n", .{
                i, jump.direction, jump.delta_s, monitor.jump_count,
            });
            try stdout.print("       Action: re-evaluate all scheduled tasks against new wall time\n", .{});
        } else {
            try stdout.print("[{d}] wall={d} - normal\n", .{ i, wall });
        }

        // sleep using monotonic time -- this is NOT affected by clock jumps
        std.time.sleep(1 * std.time.ns_per_s);
    }

    try stdout.print("\nTotal jumps detected: {d}\n", .{monitor.jump_count});
}

The monitor compares successive wall clock readings. Since we sleep using monotonic time (which std.time.sleep uses internally), we know roughly how much wall time should have elapsed. If the delta is negative (clock went backwards) or suspiciously large (clock jumped forward), we flag it. The daemon can then re-evaluate its schedule: recalculate when each cron job should next fire based on the new wall time.

This is exactly the problem that systemd's timer units handle for you -- they use a combination of monotonic and realtime clocks and automatically adjust for clock changes. But if you're building your own scheduler (embedded, BSD, no systemd), you need to handle this yourself.

Practical example: a cron-style job scheduler

Let's combine everything into a working cron scheduler. It parses cron expressions, checks them every 30 seconds against the current wall time, uses monotonic sleeps to avoid drift, and detects clock jumps:

const std = @import("std");
const linux = std.os.linux;
const posix = std.posix;

const Job = struct {
    name: []const u8,
    minute_bits: u64,  // bit set for valid minutes
    hour_bits: u64,    // bit set for valid hours
    wday_bits: u64,    // bit set for valid weekdays
    last_fired_minute: i64, // epoch minute when last fired (prevent double-fire)
    fire_count: u32,
};

fn allBits(min: u6, max: u6) u64 {
    var bits: u64 = 0;
    var v = min;
    while (v <= max) : (v += 1) {
        bits |= @as(u64, 1) << v;
        if (v == max) break;
    }
    return bits;
}

fn stepBits(min: u6, max: u6, step: u6) u64 {
    var bits: u64 = 0;
    var v: u7 = min;
    while (v <= max) {
        bits |= @as(u64, 1) << @as(u6, @intCast(v));
        v += step;
    }
    return bits;
}

fn epochToComponents(epoch: i64) struct { minute: u6, hour: u6, wday: u6, epoch_minute: i64 } {
    const es: u64 = @intCast(epoch);
    const secs_in_day: u64 = 86400;
    const days = es / secs_in_day;
    const day_secs = es % secs_in_day;
    return .{
        .minute = @intCast((day_secs % 3600) / 60),
        .hour = @intCast(day_secs / 3600),
        .wday = @intCast((days + 4) % 7), // epoch was Thursday
        .epoch_minute = @divTrunc(epoch, 60),
    };
}

pub fn main() !void {
    const stdout = std.io.getStdOut().writer();

    // define some jobs
    var jobs = [_]Job{
        .{
            .name = "every-5-min",
            .minute_bits = stepBits(0, 59, 5),
            .hour_bits = allBits(0, 23),
            .wday_bits = allBits(0, 6),
            .last_fired_minute = -1,
            .fire_count = 0,
        },
        .{
            .name = "hourly",
            .minute_bits = @as(u64, 1) << 0, // minute 0 only
            .hour_bits = allBits(0, 23),
            .wday_bits = allBits(0, 6),
            .last_fired_minute = -1,
            .fire_count = 0,
        },
        .{
            .name = "weekday-9am",
            .minute_bits = @as(u64, 1) << 0,
            .hour_bits = @as(u64, 1) << 9,
            .wday_bits = stepBits(1, 5, 1), // Mon-Fri
            .last_fired_minute = -1,
            .fire_count = 0,
        },
    };

    try stdout.print("Cron scheduler with {d} jobs. Checking every 30s...\n\n", .{jobs.len});
    var last_wall = std.time.timestamp();

    for (0..20) |tick| {
        const now = std.time.timestamp();
        const tc = epochToComponents(now);

        // detect clock jumps (>60s unexpected change)
        const wall_delta = now - last_wall;
        if (wall_delta < 0 or wall_delta > 90) {
            try stdout.print("[tick {d}] CLOCK JUMP: delta={d}s, recalculating schedules\n", .{ tick, wall_delta });
        }
        last_wall = now;

        // check each job against current time
        for (&jobs) |*job| {
            const min_match = (job.minute_bits >> tc.minute) & 1 == 1;
            const hr_match = (job.hour_bits >> tc.hour) & 1 == 1;
            const wd_match = (job.wday_bits >> tc.wday) & 1 == 1;

            if (min_match and hr_match and wd_match) {
                // prevent double-fire in the same calendar minute
                if (tc.epoch_minute != job.last_fired_minute) {
                    job.last_fired_minute = tc.epoch_minute;
                    job.fire_count += 1;
                    try stdout.print("[tick {d}] FIRE '{s}' (#{d}) at {d:0>2}:{d:0>2} wday={d}\n", .{
                        tick, job.name, job.fire_count, tc.hour, tc.minute, tc.wday,
                    });
                }
            }
        }

        if (tick < 19) {
            // sleep 30s using monotonic clock (immune to wall clock changes)
            std.time.sleep(30 * std.time.ns_per_s);
        }
    }

    try stdout.print("\nFinal job stats:\n", .{});
    for (jobs) |job| {
        try stdout.print("  {s}: fired {d} times\n", .{ job.name, job.fire_count });
    }
}

The last_fired_minute field is the key to preventing double-fires. Even if the clock jumps backwards into the same calendar minute, we won't fire the job again because the epoch-minute value was already recorded. Conversely, if the clock jumps forward skipping a minute boundary, we catch up on the next check -- the job fires at the next matching minute rather than being lost forever.

This pattern -- check cron fields using bit sets, prevent duplicates with a "last fired" marker, use monotonic sleeps between checks -- is robust against NTP adjustments, VM snapshots, and manual clock changes. The 30-second poll interval means worst-case you're ~30 seconds late on a job, which is perfectly acceptible for cron-style scheduling (real cron has the same resolution -- it checks once per minute).

Exercises

  1. Add timerfd integration to the interval scheduler from the "Building a simple interval scheduler" section. Instead of calculating sleep durations and calling std.time.sleep, create one timerfd per task (each set to that task's interval). Use a single poll() call to wait on all timerfds simultaneously, and when one becomes readable, run the corresponding task's callback. Track the actual vs expected fire times and print the jitter (difference between expected and actual) for each task. Compare the jitter to the sleep-based version -- timerfd should be significantly more consistent because the kernel manages the timing.

  2. Extend the cron parser to support comma-separated values and ranges. Comma-separated: 1,15,30 means "at minute 1, 15, and 30". Ranges: 9-17 means "hours 9 through 17 inclusive". Combined: 0,30 9-17 * * 1-5 means "at :00 and :30 during business hours on weekdays". Write tests (using std.testing) that verify parsing of at least 5 different expressions including edge cases: 59 23 31 12 0 (last minute of the year on a Sunday), */1 * * * * (every minute), and 0 0 1 1 * (midnight on January 1st). Verify both the bit sets and the matchesValue results.

  3. Implement a hierarchical timer wheel with two levels: a fine wheel (256 slots, 10ms resolution) and a coarse wheel (64 slots, 2.56s resolution = 256 * 10ms). Timers with delays up to 2.56 seconds go directly into the fine wheel. Timers with delays up to ~164 seconds (64 * 2.56s) go into the coarse wheel. When the fine wheel completes a full revolution, advance the coarse wheel by one slot and cascade its timers down into the fine wheel (recalculating their slot positions for the remaining delay). Insert 100 timers with random delays between 10ms and 60 seconds and verify they all fire within 10ms of their expected time. Print the maximum deviation observed.

Alright, zo gezegd, zo gedaan!

De groeten!

@scipio

TAGS: [ #stem ] [ #stemsocial ] [ #steemstem ] [ #zig ] [ #programming ]

Replies

NO REPLIES FOUND.

[ BACK TO TRENDING ] [ BACK TO MENU ]
CMD>