Skip to content
zit Artifact [df5046155e]
Login

Artifact [df5046155e]

Check-ins Using Download Hex

Artifact df5046155ef6bc40c3875e7aaf765057bb8ad97a5d60139976a1783497104f90:

//! The repository's write locks, taken the way git and most Unix programs
//! take them: a lock is a file created with `O_CREAT|O_EXCL`, which the
//! filesystem guarantees only one process can do, and it is released by
//! deleting the file.
//!
//! This replaces `flock` on the store file itself. `flock` is not POSIX, its
//! behaviour on network filesystems ranges from correct to silently locking
//! nothing, and a blocked `flock` waits forever without saying why. Exclusive
//! create is defined by POSIX, is atomic on NFSv3 and later, and leaves a file
//! that says who holds the lock, so a waiting process can say what it is
//! waiting for and give up with a useful message.
//!
//! What it costs is that a lock outlives a crashed holder. The file records the
//! holder's process id and host. A lock held by a process on this host that no
//! longer exists is stale, and is broken automatically; one held from another
//! host cannot be checked, so a waiter gives up and names the file to delete.
//!
//! Breaking a stale lock is itself serialised, by a second exclusive-create
//! lock beside the first, and deletes the stale file only if it still holds
//! exactly the bytes that were judged stale. Only a breaker or the holder ever
//! deletes a lock file, and a stale lock's holder is gone, so a breaker cannot
//! remove a lock some live process has just taken.
const std = @import("std");
const builtin = @import("builtin");

/// The store's writer lock, beside `.zit`. Every append takes it.
pub const writer_lock_name = ".zit.lock";

/// How long a writer waits for a live holder before giving up.
pub const default_wait_ms: u64 = 60_000;

/// A waiter says what it is waiting for once it has waited this long, so a
/// lock that is released at once never prints anything.
const announce_after_ms: u64 = 1_000;

/// A lock file with no readable holder is being written by the process that
/// just created it. Past this age its creator died between creating and
/// writing it, and it is stale.
const unreadable_stale_after_ms: u64 = 10_000;

const first_delay_ms: u64 = 5;
const max_delay_ms: u64 = 200;
const max_record = 320;
const max_name = 64;

pub const AcquireError = error{
    /// A live holder kept the lock past the wait. The message is written.
    Busy,
    /// This process already holds the lock. Waiting would wait for itself.
    HeldByThisProcess,
    Io,
};

/// A held lock. `release` deletes the file and closes the directory handle.
pub const Lock = struct {
    dir: std.Io.Dir,
    name_buf: [max_name]u8,
    name_len: usize,
    record_buf: [max_record]u8,
    record_len: usize,

    fn name(self: *const Lock) []const u8 {
        return self.name_buf[0..self.name_len];
    }

    /// Deletes the lock file if it is still ours. A lock that was wrongly
    /// broken and retaken belongs to someone else now, and is left alone.
    pub fn release(self: *Lock, io: std.Io) void {
        var buf: [max_record]u8 = undefined;
        if (self.dir.readFile(io, self.name(), &buf)) |held| {
            if (std.mem.eql(u8, held, self.record_buf[0..self.record_len])) {
                self.dir.deleteFile(io, self.name()) catch {};
            }
        } else |_| {}
        self.dir.close(io);
        self.* = undefined;
    }
};

/// Take the lock `name` in `dir`, waiting up to `wait_ms` for a live holder.
/// `wait_ms == 0` tries once. `notices` gets the one-line "waiting for" and
/// "gave up" messages; the caller flushes it.
pub fn acquire(
    io: std.Io,
    dir: std.Io.Dir,
    name: []const u8,
    wait_ms: u64,
    notices: *std.Io.Writer,
) AcquireError!Lock {
    if (name.len == 0 or name.len + ".break".len > max_name) return error.Io;
    var lock: Lock = undefined;
    @memcpy(lock.name_buf[0..name.len], name);
    lock.name_len = name.len;
    const me = Identity.current();
    lock.record_len = me.record(io, &lock.record_buf).len;

    var waited_ms: u64 = 0;
    var delay_ms: u64 = first_delay_ms;
    var announced = false;
    while (true) {
        if (tryCreate(io, dir, name, lock.record_buf[0..lock.record_len])) |created| {
            if (!created) return error.Io;
            lock.dir = dir.openDir(io, ".", .{}) catch {
                dir.deleteFile(io, name) catch {};
                return error.Io;
            };
            return lock;
        } else |err| switch (err) {
            error.Exists => {},
        }

        var held_buf: [max_record]u8 = undefined;
        const held = readHolder(io, dir, name, &held_buf) orelse continue; // released meanwhile
        switch (classify(io, held, me)) {
            .ours => {
                notices.writeAll("zit: internal error: this process already holds the repository lock\n") catch {};
                return error.HeldByThisProcess;
            },
            .stale => {
                breakStale(io, dir, name, held.bytes, me);
                // Whether this breaker won or another did, the lock is worth
                // trying again at once; only a lost race costs a short wait.
            },
            .live => {
                if (waited_ms >= wait_ms) {
                    renderBusy(notices, held, name, waited_ms);
                    return error.Busy;
                }
                if (!announced and waited_ms >= announce_after_ms) {
                    renderWaiting(notices, held);
                    notices.flush() catch {};
                    announced = true;
                }
            },
        }
        io.sleep(.fromMilliseconds(@intCast(delay_ms)), .awake) catch return error.Io;
        waited_ms += delay_ms;
        delay_ms = @min(delay_ms * 2, max_delay_ms);
    }
}

/// Who a lock record names: this process's id and host.
const Identity = struct {
    pid: i64,
    host_buf: [std.posix.HOST_NAME_MAX]u8,
    host_len: usize,

    fn current() Identity {
        var id: Identity = .{ .pid = getpid(), .host_buf = undefined, .host_len = 0 };
        const got = std.posix.gethostname(&id.host_buf) catch "";
        id.host_len = got.len;
        return id;
    }

    fn host(self: *const Identity) []const u8 {
        return self.host_buf[0..self.host_len];
    }

    /// `pid=N host=H time=T`: the time makes every record distinct, so a
    /// recycled process id never matches an older record byte for byte.
    fn record(self: *const Identity, io: std.Io, buf: *[max_record]u8) []const u8 {
        const now = std.Io.Timestamp.now(io, .real).toNanoseconds();
        return std.fmt.bufPrint(buf, "pid={d} host={s} time={d}\n", .{ self.pid, self.host(), now }) catch unreachable;
    }
};

fn getpid() i64 {
    return if (builtin.os.tag == .linux) std.os.linux.getpid() else std.c.getpid();
}

/// Whether process `pid` exists on this host. Permission to signal it is not
/// needed to know it exists.
fn processExists(pid: i64) bool {
    if (pid <= 0 or pid > std.math.maxInt(std.posix.pid_t)) return false;
    std.posix.kill(@intCast(pid), @enumFromInt(0)) catch |e| return e == error.PermissionDenied;
    return true;
}

/// true: created and the record written. false: created, but the record could
/// not be written, and the file was removed again.
///
/// The record is fsynced. Unsynced, a crash of the machine can leave the file
/// empty, and an empty lock file cannot be judged stale until it is old; with
/// the record on disk the next writer can see its holder is gone.
fn tryCreate(io: std.Io, dir: std.Io.Dir, name: []const u8, rec: []const u8) error{Exists}!bool {
    const file = dir.createFile(io, name, .{ .exclusive = true }) catch |e| switch (e) {
        error.PathAlreadyExists => return error.Exists,
        else => return false,
    };
    defer file.close(io);
    file.writePositionalAll(io, rec, 0) catch {
        dir.deleteFile(io, name) catch {};
        return false;
    };
    file.sync(io) catch {
        dir.deleteFile(io, name) catch {};
        return false;
    };
    return true;
}

const Holder = struct {
    bytes: []const u8,
    pid: ?i64,
    host: []const u8,
    age_ms: u64,
};

/// The current lock file's record, or null if there is no lock file now.
fn readHolder(io: std.Io, dir: std.Io.Dir, name: []const u8, buf: *[max_record]u8) ?Holder {
    const file = dir.openFile(io, name, .{}) catch return null;
    defer file.close(io);
    const n = file.readPositionalAll(io, buf, 0) catch return null;
    const st = file.stat(io) catch return null;
    const now = std.Io.Timestamp.now(io, .real);
    const age_ns = st.mtime.durationTo(now).toNanoseconds();
    var holder: Holder = .{
        .bytes = buf[0..n],
        .pid = null,
        .host = "",
        .age_ms = if (age_ns <= 0) 0 else @intCast(@divTrunc(age_ns, std.time.ns_per_ms)),
    };
    var fields = std.mem.tokenizeAny(u8, buf[0..n], " \n");
    while (fields.next()) |field| {
        if (std.mem.startsWith(u8, field, "pid=")) {
            holder.pid = std.fmt.parseInt(i64, field["pid=".len..], 10) catch null;
        } else if (std.mem.startsWith(u8, field, "host=")) {
            holder.host = field["host=".len..];
        }
    }
    return holder;
}

const Verdict = enum { ours, live, stale };

fn classify(io: std.Io, held: Holder, me: Identity) Verdict {
    _ = io;
    const pid = held.pid orelse
        return if (held.age_ms >= unreadable_stale_after_ms) .stale else .live;
    // Another host's processes cannot be checked from here.
    if (!std.mem.eql(u8, held.host, me.host())) return .live;
    if (pid == me.pid) return .ours;
    return if (processExists(pid)) .live else .stale;
}

/// Delete `name` if it still holds `stale` byte for byte, under the break lock.
fn breakStale(io: std.Io, dir: std.Io.Dir, name: []const u8, stale: []const u8, me: Identity) void {
    var brk_buf: [max_name]u8 = undefined;
    const brk = std.fmt.bufPrint(&brk_buf, "{s}.break", .{name}) catch unreachable;
    var rec_buf: [max_record]u8 = undefined;
    const rec = me.record(io, &rec_buf);
    if (tryCreate(io, dir, brk, rec)) |created| {
        if (!created) return;
    } else |_| {
        // Another process is breaking it. If that process died doing so, its
        // break lock is stale in turn: remove it and let the caller retry.
        var held_buf: [max_record]u8 = undefined;
        const held = readHolder(io, dir, brk, &held_buf) orelse return;
        if (classify(io, held, me) == .stale) dir.deleteFile(io, brk) catch {};
        return;
    }
    defer dir.deleteFile(io, brk) catch {};
    var now_buf: [max_record]u8 = undefined;
    const now = dir.readFile(io, name, &now_buf) catch return;
    if (std.mem.eql(u8, now, stale)) dir.deleteFile(io, name) catch {};
}

fn renderWaiting(w: *std.Io.Writer, held: Holder) void {
    if (held.pid) |pid| {
        w.print("zit: waiting for the repository lock, held by process {d} on {s}\n", .{ pid, held.host }) catch {};
    } else {
        w.writeAll("zit: waiting for the repository lock\n") catch {};
    }
}

fn renderBusy(w: *std.Io.Writer, held: Holder, name: []const u8, waited_ms: u64) void {
    const seconds = waited_ms / std.time.ms_per_s;
    if (held.pid) |pid| {
        w.print("zit: the repository is locked by process {d} on {s}", .{ pid, held.host }) catch {};
    } else {
        w.writeAll("zit: the repository is locked") catch {};
    }
    if (seconds > 0) w.print("; gave up after {d}s", .{seconds}) catch {};
    w.print(". If no zit process is using it, delete {s} beside the repository's .zit file\n", .{name}) catch {};
}

// ---------------------------------------------------------------------------

const testing = std.testing;

fn discard() std.Io.Writer.Discarding {
    return std.Io.Writer.Discarding.init(&.{});
}

test "a lock is exclusive until released, and released means the file is gone" {
    const io = testing.io;
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var sink = discard();

    var a = try acquire(io, tmp.dir, writer_lock_name, 0, &sink.writer);
    // A second acquire in the same process is refused at once, never waited on:
    // waiting would wait for itself.
    try testing.expectError(error.HeldByThisProcess, acquire(io, tmp.dir, writer_lock_name, 5_000, &sink.writer));
    a.release(io);
    try testing.expectError(error.FileNotFound, tmp.dir.access(io, writer_lock_name, .{}));
    var b = try acquire(io, tmp.dir, writer_lock_name, 0, &sink.writer);
    b.release(io);
}

test "a lock held by a process that no longer exists is broken" {
    const io = testing.io;
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var sink = discard();
    const me = Identity.current();
    // The largest pid is never a live process.
    var rec_buf: [max_record]u8 = undefined;
    const rec = try std.fmt.bufPrint(&rec_buf, "pid={d} host={s} time=1\n", .{ std.math.maxInt(std.posix.pid_t), me.host() });
    try tmp.dir.writeFile(io, .{ .sub_path = writer_lock_name, .data = rec });

    var lock = try acquire(io, tmp.dir, writer_lock_name, 0, &sink.writer);
    lock.release(io);
    try testing.expectError(error.FileNotFound, tmp.dir.access(io, writer_lock_name, .{}));
    try testing.expectError(error.FileNotFound, tmp.dir.access(io, writer_lock_name ++ ".break", .{}));
}

test "a live holder is waited for, then named when the wait runs out" {
    const io = testing.io;
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    const me = Identity.current();
    // A live process that is not this one: the parent.
    const parent: i64 = if (builtin.os.tag == .linux) std.os.linux.getppid() else std.c.getppid();
    var rec_buf: [max_record]u8 = undefined;
    const rec = try std.fmt.bufPrint(&rec_buf, "pid={d} host={s} time=1\n", .{ parent, me.host() });
    try tmp.dir.writeFile(io, .{ .sub_path = writer_lock_name, .data = rec });

    var out: [512]u8 = undefined;
    var w = std.Io.Writer.fixed(&out);
    try testing.expectError(error.Busy, acquire(io, tmp.dir, writer_lock_name, 50, &w));
    const said = w.buffered();
    var pid_buf: [32]u8 = undefined;
    try testing.expect(std.mem.indexOf(u8, said, try std.fmt.bufPrint(&pid_buf, "process {d}", .{parent})) != null);
    try testing.expect(std.mem.indexOf(u8, said, "delete .zit.lock") != null);
    // The live holder's lock is untouched.
    var held_buf: [max_record]u8 = undefined;
    try testing.expectEqualStrings(rec, try tmp.dir.readFile(io, writer_lock_name, &held_buf));
}

test "a lock from another host is never broken, only waited for" {
    const io = testing.io;
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    const rec = "pid=1 host=some-other-host.invalid time=1\n";
    try tmp.dir.writeFile(io, .{ .sub_path = writer_lock_name, .data = rec });
    var sink = discard();
    try testing.expectError(error.Busy, acquire(io, tmp.dir, writer_lock_name, 20, &sink.writer));
    var held_buf: [max_record]u8 = undefined;
    try testing.expectEqualStrings(rec, try tmp.dir.readFile(io, writer_lock_name, &held_buf));
}

test "a break only deletes the exact record judged stale" {
    const io = testing.io;
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    const me = Identity.current();
    // Judged stale, but by the time the breaker holds the break lock a live
    // process has taken the lock: that newer record must survive.
    const fresh = "pid=1 host=x time=2\n";
    try tmp.dir.writeFile(io, .{ .sub_path = writer_lock_name, .data = fresh });
    breakStale(io, tmp.dir, writer_lock_name, "pid=1 host=x time=1\n", me);
    var held_buf: [max_record]u8 = undefined;
    try testing.expectEqualStrings(fresh, try tmp.dir.readFile(io, writer_lock_name, &held_buf));
    try testing.expectError(error.FileNotFound, tmp.dir.access(io, writer_lock_name ++ ".break", .{}));
}

test "release leaves a lock that is no longer ours" {
    const io = testing.io;
    var tmp = testing.tmpDir(.{});
    defer tmp.cleanup();
    var sink = discard();
    var lock = try acquire(io, tmp.dir, writer_lock_name, 0, &sink.writer);
    const other = "pid=1 host=x time=3\n";
    try tmp.dir.writeFile(io, .{ .sub_path = writer_lock_name, .data = other });
    lock.release(io);
    var held_buf: [max_record]u8 = undefined;
    try testing.expectEqualStrings(other, try tmp.dir.readFile(io, writer_lock_name, &held_buf));
}