daemon/pidlock.bend

cswap-bend-cli@0.1.0.0 · 81 lines · 15 definitions

1import Base
2import cswap-bend-cli@0.1.0.0/fs.bend as Fs
3import cswap-bend-cli@0.1.0.0/fs/write.bend as FsW
4import cswap-bend-cli@0.1.0.0/act.bend as Act
5import cswap-bend-cli@0.1.0.0/lock.bend as Lock
6
7# One decider per backup root: `cswap auto` holds an exclusive flock on
8# <state>/daemon.lock for as long as it runs (the kernel drops it if the
9# process dies) and writes its pid to <state>/daemon.pid. A second `auto` on
10# the same root refuses with a message naming the holder.
11
12def ActM(-A: Type) -> Type:
13  Act.of(A)
14
15def ActM.pure(-A: Type, x: A) -> Act.of(A):
16  Act.pure(A, x)
17
18def ActM.bind(-A: Type, -B: Type, m: Act.of(A), f: A -> Act.of(B)) -> Act.of(B):
19  Act.bind(A, B, m, f)
20
21def lock_path(+state_dir: String) -> String:
22  state_dir ++ "/daemon.lock"
23
24def pid_path(+state_dir: String) -> String:
25  state_dir ++ "/daemon.pid"
26
27def pid_of(r: Result<&1, &1, U32 & String, U32 & (String & String)>) -> U32:
28  match r:
29    case Fail{e}:
30      0
31    case Done{(st, (out, err))}:
32      Maybe.default(&2, U32, U32.read(String.trim(out)), 0)
33
34# This process's pid: the parent of a child shell (Bend has no getpid).
35def my_pid() -> IO(U32):
36  IO.bind(Result<&1, &1, U32 & String, U32 & (String & String)>, U32, Process.run("/bin/sh", ["-c", "echo $PPID"], "", 64, 5000), r => IO.pure(U32, pid_of(r)))
37
38def holder_text(r: Fs.Res(String)) -> String:
39  match r:
40    case Done{t}:
41      String.trim(t)
42    case Fail{e}:
43      "unknown"
44
45def busy_msg(+state_dir: String, +holder: String) -> String:
46  "Another `cswap auto` is already running on this backup root (pid " ++ holder ++ "; " ++ lock_path(state_dir) ++ " is locked). Only one decider may run: stop it first."
47
48def busy_fail(-A: Type, +state_dir: String) -> Act.of(A):
49  IO.bind(Fs.Res(String), Fs.Res(A), Fs.read_file(pid_path(state_dir)), r => IO.pure(Fs.Res(A), Fail{(11, busy_msg(state_dir, holder_text(r)))}))
50
51def claim(+state_dir: String) -> Act.of(Unit):
52  do ActM<Unit>:
53    pid : U32 <- Act.lift(U32, my_pid())
54    FsW.write_atomic(pid_path(state_dir), U32.show(pid) ++ "\n", "600")
55
56def give_back(+state_dir: String, f: File) -> Act.of(Unit):
57  do ActM<Unit>:
58    Act.lift(Unit, Lock.release(f))
59    FsW.remove(pid_path(state_dir))
60
61def run_held(-A: Type, +state_dir: String, f: File, c: Fs.Res(Unit), act: Act.of(A)) -> Act.of(A):
62  match c:
63    case Fail{e}:
64      IO.bind(Unit, Fs.Res(A), Lock.release(f), u => IO.pure(Fs.Res(A), Fail{e}))
65    case Done{u}:
66      Act.finally(A, act, give_back(state_dir, f))
67
68def held(-A: Type, +state_dir: String, m: Maybe<&1, File>, act: Act.of(A)) -> Act.of(A):
69  match m:
70    case None{}:
71      busy_fail(A, state_dir)
72    case Some{f}:
73      IO.bind(Fs.Res(Unit), Fs.Res(A), claim(state_dir), c => run_held(A, state_dir, f, c, act))
74
75# Run `act` as the one decider of this state dir. A second caller gets
76# Fail{(11, "Another `cswap auto` is already running ...")} and `act` never starts.
77def with_pidlock(-A: Type, +state_dir: String, act: Act.of(A)) -> Act.of(A):
78  do ActM<A>:
79    FsW.mkdir_p(state_dir)
80    m : Maybe<&1, File> <- Lock.acquire(lock_path(state_dir), 0)
81    held(A, state_dir, m, act)