lock.bend

cswap-bend-cli@0.1.0.0 · 110 lines · 17 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
4
5# An exclusive advisory lock that interoperates with the Python cswap's
6# locking.py: flock(LOCK_EX) on a file opened for writing at the same path,
7# so both implementations can run side by side. The lock belongs to the open
8# file; closing the File (release) drops it, and so does the process dying.
9#
10# Like the reference, a held lock is polled every 100 ms (LOCK_NB) until the
11# timeout passes, because a blocking flock could not be given up on.
12
13# One LOCK_EX|LOCK_NB attempt on the File's descriptor: Done{1} held, Done{0}
14# someone else holds it. The File comes back beside the answer.
15def Flock.try(file: File) -> IO(File & Result<&1, &1, U32 & String, U32>):
16  import "./effs/flock.c"
17  import "./effs/flock.js"
18
19def poll_ms() -> U32:
20  100
21
22def tried.bool(r: Fs.Res(U32)) -> Fs.Res(Bool):
23  match r:
24    case Fail{e}:
25      Fail{e}
26    case Done{n}:
27      Done{U32.is_eq(n, 1)}
28
29def tried(fr: File & Fs.Res(U32)) -> IO(File & Fs.Res(Bool)):
30  (f, r) = fr
31  IO.pure(File & Fs.Res(Bool), (f, tried.bool(r)))
32
33# One attempt, as a Bool.
34def attempt(f: File) -> IO(File & Fs.Res(Bool)):
35  IO.bind(File & Fs.Res(U32), File & Fs.Res(Bool), Flock.try(f), tried)
36
37# Retry `fuel` more times, 100 ms apart, until the lock is held.
38def wait(fuel: Nat, fr: File & Fs.Res(Bool)) -> IO(File & Fs.Res(Bool)):
39  match fuel fr:
40    case 1n++p Tuple{f, Done{False{}}}:
41      IO.bind(Unit, File & Fs.Res(Bool), IO.sleep(poll_ms()),
42        u => IO.bind(File & Fs.Res(Bool), File & Fs.Res(Bool), attempt(f), r => wait(p, r)))
43    case _ _:
44      IO.pure(File & Fs.Res(Bool), fr)
45
46def held.pick(f: File, r: Fs.Res(Bool)) -> IO(Fs.Res(Maybe<&1, File>)):
47  match r:
48    case Fail{e}:
49      IO.bind(Unit, Fs.Res(Maybe<&1, File>), File.close(f), u => IO.pure(Fs.Res(Maybe<&1, File>), Fail{e}))
50    case Done{True{}}:
51      IO.pure(Fs.Res(Maybe<&1, File>), Done{Some{f}})
52    case Done{False{}}:
53      IO.bind(Unit, Fs.Res(Maybe<&1, File>), File.close(f), u => IO.pure(Fs.Res(Maybe<&1, File>), Done{None{}}))
54
55def held.fin(fr: File & Fs.Res(Bool)) -> IO(Fs.Res(Maybe<&1, File>)):
56  (f, r) = fr
57  held.pick(f, r)
58
59# Attempts that fit in the timeout: the first one, then one per 100 ms.
60def tries(timeout_ms: U32) -> Nat:
61  U32.to_nat((timeout_ms / poll_ms() : U32))
62
63def opened(timeout_ms: U32, r: Fs.Res(File)) -> IO(Fs.Res(Maybe<&1, File>)):
64  match r:
65    case Fail{e}:
66      IO.pure(Fs.Res(Maybe<&1, File>), Fail{e})
67    case Done{f}:
68      IO.bind(File & Fs.Res(Bool), Fs.Res(Maybe<&1, File>),
69        IO.bind(File & Fs.Res(Bool), File & Fs.Res(Bool), attempt(f), r1 => wait(tries(timeout_ms), r1)),
70        held.fin)
71
72def parent.ok(r: Fs.Res(Unit), timeout_ms: U32, path: String) -> IO(Fs.Res(Maybe<&1, File>)):
73  match r:
74    case Fail{e}:
75      IO.pure(Fs.Res(Maybe<&1, File>), Fail{e})
76    case Done{u}:
77      IO.bind(Fs.Res(File), Fs.Res(Maybe<&1, File>), File.open(path, "w"), g => opened(timeout_ms, g))
78
79# Take the lock at `path`, waiting up to `timeout_ms`. Done{Some{file}} holds
80# it (pass the file to `release`); Done{None} means the wait timed out.
81# Creates missing parent directories and truncates the lock file, as the
82# reference does.
83def acquire(+path: String, timeout_ms: U32) -> IO(Fs.Res(Maybe<&1, File>)):
84  IO.bind(Fs.Res(Unit), Fs.Res(Maybe<&1, File>), FsW.mkdir_p(Fs.dirname(path)),
85    r => parent.ok(r, timeout_ms, path))
86
87# Drop the lock.
88def release(f: File) -> IO(Unit):
89  File.close(f)
90
91def busy() -> Fs.Res(Unit):
92  Fail{(11, "Failed to acquire lock - another instance may be running")}
93
94def done.close(-A: Type, r: Fs.Res(A), f: File) -> IO(Fs.Res(A)):
95  IO.bind(Unit, Fs.Res(A), File.close(f), u => IO.pure(Fs.Res(A), r))
96
97def held.run(-A: Type, act: IO(Fs.Res(A)), m: Fs.Res(Maybe<&1, File>)) -> IO(Fs.Res(A)):
98  match m:
99    case Fail{e}:
100      IO.pure(Fs.Res(A), Fail{e})
101    case Done{None{}}:
102      IO.pure(Fs.Res(A), Fail{(11, "Failed to acquire lock - another instance may be running")})
103    case Done{Some{f}}:
104      IO.bind(Fs.Res(A), Fs.Res(A), act, r => done.close(A, r, f))
105
106# Run `act` while holding the lock; the lock is released afterwards whatever
107# `act` answers. A timeout is Fail{(11, "Failed to acquire lock ...")}.
108def with_lock(-A: Type, path: String, timeout_ms: U32, act: IO(Fs.Res(A))) -> IO(Fs.Res(A)):
109  IO.bind(Fs.Res(Maybe<&1, File>), Fs.Res(A), acquire(path, timeout_ms), m => held.run(A, act, m))
110