//! 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), @fromBackingInt(@intCast(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));
}