src/DbLock.bend

mylsm-lsm-store@0.6.0.1 · 381 lines · 31 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/DurableError.bend as DurableError
3import mylsm-lsm-store@0.6.0.1/src/DbLockPolicy.bend as Policy
4import mylsm-lsm-store@0.6.0.1/src/Fs.bend as Fs
5
6# Owns the open LOCK file until release.
7type Lock is Type:
8  Lock{file: File}
9
10# Determines whether a clock sample selects backoff or deadline validation.
11type WaitPurpose is Data:
12  ChooseDelay{}
13  CheckDeadline{}
14
15# Carries the raw result of one nonblocking lock attempt.
16type AttemptDecision is Type:
17  Contended{}
18  Failed{error: U32 & String}
19
20# Carries file ownership through the single recursive wait state machine.
21type WaitState is Type:
22  TryLock{file: File, started_us: Nat, timeout_ms: Nat, backoff_ms: U32,
23    operation: String, path: String}
24  LockAttempt{pair: File & Result<&1, &1, U32 & String, Unit>,
25    started_us: Nat, timeout_ms: Nat, backoff_ms: U32, operation: String, path: String}
26  ReadClock{file: File, started_us: Nat, timeout_ms: Nat, backoff_ms: U32,
27    operation: String, path: String, purpose: WaitPurpose}
28  ClockSample{file: File, started_us: Nat, timeout_ms: Nat, backoff_ms: U32,
29    operation: String, path: String, purpose: WaitPurpose,
30    result: Result<&1, &1, U32 & String, Nat>}
31  Pause{file: File, started_us: Nat, timeout_ms: Nat, backoff_ms: U32,
32    operation: String, path: String, duration_ms: U32}
33  PauseResult{file: File, started_us: Nat, timeout_ms: Nat, backoff_ms: U32,
34    operation: String, path: String,
35    result: Result<&1, &1, U32 & String, Unit>}
36  WaitFatal{file: File, error: U32 & String, operation: String, path: String}
37  WaitExpired{file: File, operation: String, path: String}
38
39
40# Builds the stable LOCK path for a database directory.
41def lock_path(path: String) -> String:
42  path ++ "/LOCK"
43
44def DbLock.try_lock(file: File) -> IO(File & Result<&1, &1, U32 & String, Unit>):
45  import "./effs/db_lock.c"
46  import "./effs/db_lock.js"
47
48def DbLock.try_unlock(file: File) -> IO(File & Result<&1, &1, U32 & String, Unit>):
49  import "./effs/db_unlock.c"
50  import "./effs/db_unlock.js"
51
52# Classifies a failure to open the LOCK file.
53def open_failed(error: U32 & String, operation: String, path: String) -> IO(Result<&1, &1, DurableError.Error, Lock>):
54  match error:
55    case (code, message):
56      IO.pure(Result<&1, &1, DurableError.Error, Lock>,
57        Fail{Policy.classify_lock_error(code, message, operation, path)})
58
59# Closes the file and classifies a failed lock attempt.
60def lock_failed(
61  file: File,
62  error: U32 & String,
63  operation: String,
64  path: String
65) -> IO(Result<&1, &1, DurableError.Error, Lock>):
66  match error:
67    case (code, message):
68      do IO<Result<&1, &1, DurableError.Error, Lock>>:
69        _closed : Unit <- File.close(file)
70        return Fail{Policy.classify_lock_error(code, message, operation, path)}
71
72# Returns an owned lock on success or closes the failed handle.
73def lock_result(
74  pair: File & Result<&1, &1, U32 & String, Unit>,
75  operation: String,
76  path: String
77) -> IO(Result<&1, &1, DurableError.Error, Lock>):
78  match pair:
79    case (file, Fail{error}):
80      lock_failed(file, error, operation, path)
81    case (file, Done{Unit{}}):
82      IO.pure(Result<&1, &1, DurableError.Error, Lock>, Done{Lock{file}})
83
84# Attempts the nonblocking OS lock on the open file.
85def lock_file(file: File, operation: String, path: String) -> IO(Result<&1, &1, DurableError.Error, Lock>):
86  do IO<Result<&1, &1, DurableError.Error, Lock>>:
87    pair : File & Result<&1, &1, U32 & String, Unit> <- DbLock.try_lock(file)
88    lock_result(pair, operation, path)
89
90# Locks the opened file or classifies its open error.
91def opened(
92  result: Result<&1, &1, U32 & String, File>,
93  operation: String,
94  path: String
95) -> IO(Result<&1, &1, DurableError.Error, Lock>):
96  match result:
97    case Fail{error}:
98      open_failed(error, operation, path)
99    case Done{file}:
100      lock_file(file, operation, path)
101
102# Opens the stable LOCK file and acquires its OS lock.
103def acquire(+path: String, operation: String) -> IO(Result<&1, &1, DurableError.Error, Lock>):
104  do IO<Result<&1, &1, DurableError.Error, Lock>>:
105    result : Result<&1, &1, U32 & String, File> <- File.open(lock_path(path), "a")
106    opened(result, operation, path)
107
108# Closes the file and preserves the unlock error.
109def release_failed(
110  file: File,
111  error: U32 & String,
112  operation: String,
113  path: String
114) -> IO(Result<&1, &1, DurableError.Error, Unit>):
115  match error:
116    case (code, message):
117      do IO<Result<&1, &1, DurableError.Error, Unit>>:
118        _closed : Unit <- File.close(file)
119        return Fail{DurableError.classify_open(DurableError.OpenHostFailure{code, message}, operation, path)}
120
121# Closes the file after a successful unlock.
122def release_succeeded(file: File) -> IO(Result<&1, &1, DurableError.Error, Unit>):
123  do IO<Result<&1, &1, DurableError.Error, Unit>>:
124    _closed : Unit <- File.close(file)
125    return Done{Unit{}}
126
127# Closes the file after either unlock result.
128def released(
129  pair: File & Result<&1, &1, U32 & String, Unit>,
130  operation: String,
131  path: String
132) -> IO(Result<&1, &1, DurableError.Error, Unit>):
133  match pair:
134    case (file, Fail{error}):
135      release_failed(file, error, operation, path)
136    case (file, Done{Unit{}}):
137      release_succeeded(file)
138
139# Unlocks the file and closes it after the result.
140def unlock_file(file: File, operation: String, path: String) -> IO(Result<&1, &1, DurableError.Error, Unit>):
141  do IO<Result<&1, &1, DurableError.Error, Unit>>:
142    pair : File & Result<&1, &1, U32 & String, Unit> <- DbLock.try_unlock(file)
143    released(pair, operation, path)
144
145# Consumes the lock and releases its file handle.
146def release(lock: Lock, operation: String, path: String) -> IO(Result<&1, &1, DurableError.Error, Unit>):
147  match lock:
148    case Lock{file}:
149      unlock_file(file, operation, path)
150
151# Closes the lock file and returns a distinct timeout error.
152def wait.timeout(file: File, operation: String, path: String) -> IO(Result<&1, &1, DurableError.Error, Lock>):
153  do IO<Result<&1, &1, DurableError.Error, Lock>>:
154    _closed : Unit <- File.close(file)
155    return Fail{DurableError.Error{DurableError.Timeout{}, operation, path, 0,
156      "lock acquisition timed out"}}
157
158# Closes the lock file and classifies a clock or sleep host failure.
159def wait.host_error(
160  file: File,
161  error: U32 & String,
162  operation: String,
163  path: String
164) -> IO(Result<&1, &1, DurableError.Error, Lock>):
165  match error:
166    case (code, message):
167      do IO<Result<&1, &1, DurableError.Error, Lock>>:
168        _closed : Unit <- File.close(file)
169        return Fail{DurableError.classify_open(
170          DurableError.OpenHostFailure{code, message}, operation, path)}
171
172# Separates contention from fatal lock errors while preserving host details.
173def wait.attempt.decision(error: U32 & String) -> AttemptDecision:
174  match error:
175    case (+code, message):
176      Bool.pick(AttemptDecision, Policy.busy_code(code),
177        Contended{}, Failed{(code, message)})
178
179# Converts a raw lock failure into the next owned state.
180def wait.attempt.state(
181  decision: AttemptDecision,
182  file: File,
183  started_us: Nat,
184  timeout_ms: Nat,
185  backoff_ms: U32,
186  operation: String,
187  path: String
188) -> WaitState:
189  match decision:
190    case Contended{}:
191      ReadClock{file, started_us, timeout_ms, backoff_ms,
192        operation, path, ChooseDelay{}}
193    case Failed{error}:
194      WaitFatal{file, error, operation, path}
195
196# Converts backoff policy into either one sleep or a terminal state.
197def wait.delay.state(
198  decision: Policy.WaitDecision,
199  file: File,
200  started_us: Nat,
201  timeout_ms: Nat,
202  backoff_ms: U32,
203  operation: String,
204  path: String
205) -> WaitState:
206  match decision:
207    case Policy.LockRetry{sleep_ms}:
208      Pause{file, started_us, timeout_ms, backoff_ms, operation, path,
209        U32.from_nat(sleep_ms)}
210    case _:
211      WaitExpired{file, operation, path}
212
213# Converts a post-sleep deadline sample into a retry or timeout state.
214def wait.deadline.state(
215  remaining_ms: Nat,
216  file: File,
217  started_us: Nat,
218  timeout_ms: Nat,
219  backoff_ms: U32,
220  operation: String,
221  path: String
222) -> WaitState:
223  match remaining_ms:
224    case 0n:
225      WaitExpired{file, operation, path}
226    case 1n+_:
227      TryLock{file, started_us, timeout_ms, backoff_ms, operation, path}
228
229# Selects backoff or deadline logic from a monotonic elapsed-time sample.
230def wait.clock.decision(
231  purpose: WaitPurpose,
232  file: File,
233  started_us: Nat,
234  +timeout_ms: Nat,
235  +backoff_ms: U32,
236  operation: String,
237  path: String,
238  elapsed_ms: Nat
239) -> WaitState:
240  match purpose:
241    case ChooseDelay{}:
242      wait.delay.state(Policy.busy_wait_decision(elapsed_ms, timeout_ms,
243        U32.to_nat(backoff_ms)), file, started_us, timeout_ms, backoff_ms,
244        operation, path)
245    case CheckDeadline{}:
246      wait.deadline.state(Nat.sub(timeout_ms, elapsed_ms), file, started_us,
247        timeout_ms, backoff_ms, operation, path)
248
249# Closes a handle if the finite retry budget is unexpectedly exhausted.
250def wait.exhausted(state: WaitState) -> IO(Result<&1, &1, DurableError.Error, Lock>):
251  match state:
252    case TryLock{file, _, _, _, operation, path}:
253      wait.timeout(file, operation, path)
254    case LockAttempt{pair, _, _, _, operation, path}:
255      match pair:
256        case (file, _):
257          wait.timeout(file, operation, path)
258    case ReadClock{file, _, _, _, operation, path, _}:
259      wait.timeout(file, operation, path)
260    case ClockSample{file, _, _, _, operation, path, _, _}:
261      wait.timeout(file, operation, path)
262    case Pause{file, _, _, _, operation, path, _}:
263      wait.timeout(file, operation, path)
264    case PauseResult{file, _, _, _, operation, path, _}:
265      wait.timeout(file, operation, path)
266    case WaitFatal{file, _, operation, path}:
267      wait.timeout(file, operation, path)
268    case WaitExpired{file, operation, path}:
269      wait.timeout(file, operation, path)
270
271# Runs lock, clock, and sleep transitions with one decreasing structural fuel.
272def wait.loop(fuel: Nat, state: WaitState) -> IO(Result<&1, &1, DurableError.Error, Lock>):
273  match fuel:
274    case 0n:
275      wait.exhausted(state)
276    case 1n+rest:
277      match state:
278        case TryLock{file, started_us, timeout_ms, backoff_ms, operation, path}:
279          do IO<Result<&1, &1, DurableError.Error, Lock>>:
280            pair : File & Result<&1, &1, U32 & String, Unit> <- DbLock.try_lock(file)
281            wait.loop(rest, LockAttempt{pair, started_us, timeout_ms,
282              backoff_ms, operation, path})
283        case LockAttempt{pair, started_us, timeout_ms, backoff_ms, operation, path}:
284          match pair:
285            case (file, Done{Unit{}}):
286              IO.pure(Result<&1, &1, DurableError.Error, Lock>, Done{Lock{file}})
287            case (file, Fail{error}):
288              wait.loop(rest, wait.attempt.state(wait.attempt.decision(error),
289                file, started_us, timeout_ms, backoff_ms, operation, path))
290        case ReadClock{file, started_us, timeout_ms, backoff_ms, operation, path, purpose}:
291          do IO<Result<&1, &1, DurableError.Error, Lock>>:
292            result : Result<&1, &1, U32 & String, Nat> <- Fs.clock_monotonic()
293            wait.loop(rest, ClockSample{file, started_us, timeout_ms,
294              backoff_ms, operation, path, purpose, result})
295        case ClockSample{file, +started_us, +timeout_ms, +backoff_ms,
296            operation, path, purpose, result}:
297          match result:
298            case Fail{error}:
299              wait.host_error(file, error, operation, path)
300            case Done{now_us}:
301              +elapsed_us = Nat.sub(now_us, started_us)
302              +elapsed_ms = Nat.div(elapsed_us, 1000n)
303              wait.loop(rest, wait.clock.decision(purpose, file, started_us,
304                timeout_ms, backoff_ms, operation, path, elapsed_ms))
305        case Pause{file, started_us, timeout_ms, backoff_ms, operation, path, duration_ms}:
306          do IO<Result<&1, &1, DurableError.Error, Lock>>:
307            result : Result<&1, &1, U32 & String, Unit> <- Fs.sleep_ms(duration_ms)
308            wait.loop(rest, PauseResult{file, started_us, timeout_ms,
309              backoff_ms, operation, path, result})
310        case PauseResult{file, started_us, timeout_ms, backoff_ms, operation, path, result}:
311          match result:
312            case Fail{error}:
313              wait.host_error(file, error, operation, path)
314            case Done{Unit{}}:
315              wait.loop(rest, ReadClock{file, started_us, timeout_ms,
316                backoff_ms, operation, path, CheckDeadline{}})
317        case WaitFatal{file, error, operation, path}:
318          lock_failed(file, error, operation, path)
319        case WaitExpired{file, operation, path}:
320          wait.timeout(file, operation, path)
321
322# Begins retries after validating the monotonic clock source.
323def acquire_wait.started(
324  file: File,
325  operation: String,
326  path: String,
327  timeout_ms: U32,
328  backoff_ms: U32,
329  result: Result<&1, &1, U32 & String, Nat>
330) -> IO(Result<&1, &1, DurableError.Error, Lock>):
331  match result:
332    case Fail{error}:
333      wait.host_error(file, error, operation, path)
334    case Done{started_us}:
335      +timeout = U32.to_nat(timeout_ms)
336      fuel = Nat.add(Nat.mul(timeout, 8n), 8n)
337      wait.loop(fuel, TryLock{file, started_us, timeout, backoff_ms,
338        operation, path})
339
340# Opens the stable lock file and samples the monotonic start time.
341def acquire_wait.opened(
342  result: Result<&1, &1, U32 & String, File>,
343  operation: String,
344  path: String,
345  timeout_ms: U32,
346  backoff_ms: U32
347) -> IO(Result<&1, &1, DurableError.Error, Lock>):
348  match result:
349    case Fail{error}:
350      open_failed(error, operation, path)
351    case Done{file}:
352      do IO<Result<&1, &1, DurableError.Error, Lock>>:
353        started : Result<&1, &1, U32 & String, Nat> <- Fs.clock_monotonic()
354        acquire_wait.started(file, operation, path, timeout_ms, backoff_ms, started)
355
356# Opens and waits only after the pure option validation succeeds.
357def acquire_wait.valid(
358  valid: Bool,
359  +path: String,
360  operation: String,
361  +timeout_ms: U32,
362  +backoff_ms: U32
363) -> IO(Result<&1, &1, DurableError.Error, Lock>):
364  match valid:
365    case False{}:
366      IO.pure(Result<&1, &1, DurableError.Error, Lock>,
367        Fail{DurableError.classify_open(DurableError.InvalidInput{}, operation, path)})
368    case True{}:
369      do IO<Result<&1, &1, DurableError.Error, Lock>>:
370        opened : Result<&1, &1, U32 & String, File> <- File.open(lock_path(path), "a")
371        acquire_wait.opened(opened, operation, path, timeout_ms, backoff_ms)
372
373# Acquires the lock immediately or waits with a finite monotonic timeout.
374def acquire_wait(
375  +path: String,
376  operation: String,
377  +timeout_ms: U32,
378  +backoff_ms: U32
379) -> IO(Result<&1, &1, DurableError.Error, Lock>):
380  acquire_wait.valid(Policy.wait_options_valid(timeout_ms, backoff_ms),
381    path, operation, timeout_ms, backoff_ms)