src/CompactIo.bend
mylsm-lsm-store@0.5.0.0 · 419 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.5.0.0/src/Compact.bend as Compact
3import mylsm-lsm-store@0.5.0.0/src/MemTable.bend as MemTable
4import mylsm-lsm-store@0.5.0.0/src/Sstable.bend as Sstable
5import mylsm-lsm-store@0.5.0.0/src/SstStreamIo.bend as SstStreamIo
6import mylsm-lsm-store@0.5.0.0/src/Wal.bend as Wal
7import mylsm-lsm-store@0.5.0.0/src/Manifest.bend as Manifest
8import mylsm-lsm-store@0.5.0.0/src/Flush.bend as Flush
9import mylsm-lsm-store@0.5.0.0/src/Fs.bend as Fs
10import mylsm-lsm-store@0.5.0.0/src/Db.bend as Db
11import mylsm-lsm-store@0.5.0.0/src/CrashPoint.bend as CrashPoint
12
13# Handle remove list in the compaction filesystem effects.
14def remove_list(names: List<&2, String>) -> IO(Result<&1, &1, U32 & String, Unit>):
15 match names:
16 case Nil{}:
17 IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
18 case Con{n, t}:
19 do IO<Result<&1, &1, U32 & String, Unit>>:
20 _rm : Result<&1, &1, U32 & String, Unit> <- Fs.remove(n)
21 remove_list(t)
22
23# Tracks progress while removing published files.
24type RemoveState is Type:
25 RemoveStart{names: List<&2, String>}
26 RemoveNames{names: List<&2, String>}
27 RemoveResult{result: Result<&1, &1, U32 & String, Unit>, names: List<&2, String>}
28
29# Removes every path and propagates the first failure.
30def remove_list_strict(fuel: Nat, state: RemoveState) -> IO(Result<&1, &1, U32 & String, Unit>):
31 match fuel:
32 case 0n:
33 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{(U32.from_nat(7n), "remove list exhausted")})
34 case 1n+rest:
35 match state:
36 case RemoveStart{names}:
37 remove_list_strict(rest, RemoveNames{names})
38 case RemoveNames{names}:
39 match names:
40 case Nil{}:
41 IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
42 case Con{name, tail}:
43 do IO<Result<&1, &1, U32 & String, Unit>>:
44 removed : Result<&1, &1, U32 & String, Unit> <- Fs.remove(name)
45 remove_list_strict(rest, RemoveResult{removed, tail})
46 case RemoveResult{result, names}:
47 match result:
48 case Fail{error}:
49 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
50 case Done{Unit{}}:
51 remove_list_strict(rest, RemoveNames{names})
52
53# Starts checked removal of the published file list.
54def remove_list_strict_start(pair: Nat & List<&2, String>) -> IO(Result<&1, &1, U32 & String, Unit>):
55 match pair:
56 case (fuel, names):
57 remove_list_strict(fuel, RemoveStart{names})
58
59def compact_goB.removed(
60 result: Result<&1, &1, U32 & String, Unit>,
61 +dir: String,
62 +mem: MemTable.MemTable,
63 +levels2: List<&2, List<&2, Sstable.Table>>,
64 +flushed: Nat,
65 +manifest: Manifest.Manifest,
66 +frozen: MemTable.MemTable,
67 +batch_cap: Nat,
68 +mem_count: Nat,
69 +frozen_count: Nat
70) -> IO(Result<&1, &1, U32 & String, Db.Db>):
71 match result:
72 case Fail{error}:
73 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
74 case Done{Unit{}}:
75 IO.pure(Result<&1, &1, U32 & String, Db.Db>,
76 Done{Db.Db{dir, mem, frozen, batch_cap, Nil{}, levels2,
77 Nat.add(flushed, 1n), Manifest.token(manifest), mem_count, frozen_count}})
78
79def compact_goB.directory_synced(
80 result: Result<&1, &1, U32 & String, Unit>,
81 +dir: String,
82 +dels: List<&2, String>,
83 +m2: Manifest.Manifest,
84 +mem: MemTable.MemTable,
85 +levels2: List<&2, List<&2, Sstable.Table>>,
86 +flushed: Nat,
87 +frozen: MemTable.MemTable,
88 +batch_cap: Nat,
89 +mem_count: Nat,
90 +frozen_count: Nat
91) -> IO(Result<&1, &1, U32 & String, Db.Db>):
92 match result:
93 case Fail{error}:
94 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
95 case Done{Unit{}}:
96 do IO<Result<&1, &1, U32 & String, Db.Db>>:
97 _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("compact.manifest_published"))
98 removed : Result<&1, &1, U32 & String, Unit> <- remove_list_strict_start((Nat.add(Nat.mul(2n, List.length(&2, String, dels)), 2n), dels))
99 compact_goB.removed(removed, dir, mem, levels2, flushed, m2,
100 frozen, batch_cap, mem_count, frozen_count)
101
102def compact_goB.renamed(
103 result: Result<&1, &1, U32 & String, Unit>,
104 +dir: String,
105 +dels: List<&2, String>,
106 +m2: Manifest.Manifest,
107 +mem: MemTable.MemTable,
108 +levels2: List<&2, List<&2, Sstable.Table>>,
109 +flushed: Nat,
110 +frozen: MemTable.MemTable,
111 +batch_cap: Nat,
112 +mem_count: Nat,
113 +frozen_count: Nat
114) -> IO(Result<&1, &1, U32 & String, Db.Db>):
115 match result:
116 case Fail{error}:
117 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
118 case Done{Unit{}}:
119 do IO<Result<&1, &1, U32 & String, Db.Db>>:
120 synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(dir)
121 compact_goB.directory_synced(synced, dir, dels, m2, mem, levels2,
122 flushed, frozen, batch_cap, mem_count, frozen_count)
123
124def compact_goB.written(
125 result: Result<&1, &1, U32 & String, Unit>,
126 +dir: String,
127 +mpath: String,
128 +mtmp: String,
129 +dels: List<&2, String>,
130 +m2: Manifest.Manifest,
131 +mem: MemTable.MemTable,
132 +levels2: List<&2, List<&2, Sstable.Table>>,
133 +flushed: Nat,
134 +frozen: MemTable.MemTable,
135 +batch_cap: Nat,
136 +mem_count: Nat,
137 +frozen_count: Nat
138) -> IO(Result<&1, &1, U32 & String, Db.Db>):
139 match result:
140 case Fail{error}:
141 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
142 case Done{Unit{}}:
143 do IO<Result<&1, &1, U32 & String, Db.Db>>:
144 _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("compact.manifest_synced"))
145 renamed : Result<&1, &1, U32 & String, Unit> <- Fs.rename(mtmp, mpath)
146 compact_goB.renamed(renamed, dir, dels, m2, mem,
147 levels2, flushed, frozen, batch_cap, mem_count, frozen_count)
148
149# Compact goB for the compaction filesystem effects.
150def compact_goB(
151 +dir: String,
152 +mpath: String,
153 +mtmp: String,
154 +m2: Manifest.Manifest,
155 +dels: List<&2, String>,
156 +mem: MemTable.MemTable,
157 +levels2: List<&2, List<&2, Sstable.Table>>,
158 +flushed: Nat,
159 +frozen: MemTable.MemTable,
160 +batch_cap: Nat,
161 +mem_count: Nat,
162 +frozen_count: Nat,
163) -> IO(Result<&1, &1, U32 & String, Db.Db>):
164 do IO<Result<&1, &1, U32 & String, Db.Db>>:
165 written : Result<&1, &1, U32 & String, Unit> <- Flush.write_manifest(mtmp, m2)
166 compact_goB.written(written, dir, mpath, mtmp, dels, m2, mem, levels2,
167 flushed, frozen, batch_cap, mem_count, frozen_count)
168
169# Handle drift go in the compaction filesystem effects.
170def drift_go(
171 ok: Bool,
172 +dir: String,
173 +mpath: String,
174 +mtmp: String,
175 +mfst: Manifest.Manifest,
176 +oldl1: List<&2, Sstable.Table>,
177 +remt: List<&2, Sstable.Table>,
178 +outname: String,
179 +mem: MemTable.MemTable,
180 +levels2: List<&2, List<&2, Sstable.Table>>,
181 +flushed: Nat,
182 +frozen: MemTable.MemTable,
183 +batch_cap: Nat,
184 +mem_count: Nat,
185 +frozen_count: Nat,
186) -> IO(Result<&1, &1, U32 & String, Db.Db>):
187 match ok:
188 case True{}:
189 +new_mfst = Compact.mfst_new(mfst, Compact.rem_names(oldl1, Compact.ml1(mfst), remt), outname)
190 +removed = Compact.abs_names(oldl1, Compact.ml1(mfst), remt)
191 +dels = List.append(&2, String, Compact.prefix(Compact.ml0(mfst), dir ++ "/l0"), Compact.prefix(removed, dir ++ "/l1"))
192 compact_goB(dir, mpath, mtmp, new_mfst, dels, mem, levels2, flushed, frozen, batch_cap, mem_count, frozen_count)
193 case False{}:
194 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(2n), "manifest drift")})
195
196def compact_goA.loaded(
197 result: Result<&1, &1, U32 & String, Manifest.Manifest>,
198 +dir: String,
199 +mpath: String,
200 +mtmp: String,
201 +oldl1: List<&2, Sstable.Table>,
202 +remt: List<&2, Sstable.Table>,
203 +outname: String,
204 +mem: MemTable.MemTable,
205 +levels2: List<&2, List<&2, Sstable.Table>>,
206 +flushed: Nat,
207 +manifest_token: String,
208 +frozen: MemTable.MemTable,
209 +batch_cap: Nat,
210 +mem_count: Nat,
211 +frozen_count: Nat
212) -> IO(Result<&1, &1, U32 & String, Db.Db>):
213 match result:
214 case Fail{error}:
215 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
216 case Done{+manifest}:
217 drift_go(String.eq(Manifest.token(manifest), manifest_token), dir, mpath,
218 mtmp, manifest, oldl1, remt, outname, mem, levels2, flushed,
219 frozen, batch_cap, mem_count, frozen_count)
220
221def compact_goA.synced(
222 result: Result<&1, &1, U32 & String, Unit>,
223 +dir: String,
224 +mpath: String,
225 +mtmp: String,
226 +oldl1: List<&2, Sstable.Table>,
227 +remt: List<&2, Sstable.Table>,
228 +outname: String,
229 +mem: MemTable.MemTable,
230 +levels2: List<&2, List<&2, Sstable.Table>>,
231 +flushed: Nat,
232 +manifest_token: String,
233 +frozen: MemTable.MemTable,
234 +batch_cap: Nat,
235 +mem_count: Nat,
236 +frozen_count: Nat
237) -> IO(Result<&1, &1, U32 & String, Db.Db>):
238 match result:
239 case Fail{error}:
240 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
241 case Done{Unit{}}:
242 do IO<Result<&1, &1, U32 & String, Db.Db>>:
243 _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("compact.output_published"))
244 manifest : Result<&1, &1, U32 & String, Manifest.Manifest> <- Flush.load_mfst(mpath)
245 compact_goA.loaded(manifest, dir, mpath, mtmp, oldl1, remt, outname,
246 mem, levels2, flushed, manifest_token, frozen, batch_cap, mem_count, frozen_count)
247
248def compact_goA.renamed(
249 result: Result<&1, &1, U32 & String, Unit>,
250 +dir: String,
251 +l1dir: String,
252 +mpath: String,
253 +mtmp: String,
254 +oldl1: List<&2, Sstable.Table>,
255 +remt: List<&2, Sstable.Table>,
256 +outname: String,
257 +mem: MemTable.MemTable,
258 +levels2: List<&2, List<&2, Sstable.Table>>,
259 +flushed: Nat,
260 +manifest_token: String,
261 +frozen: MemTable.MemTable,
262 +batch_cap: Nat,
263 +mem_count: Nat,
264 +frozen_count: Nat
265) -> IO(Result<&1, &1, U32 & String, Db.Db>):
266 match result:
267 case Fail{error}:
268 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
269 case Done{Unit{}}:
270 do IO<Result<&1, &1, U32 & String, Db.Db>>:
271 synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(l1dir)
272 compact_goA.synced(synced, dir, mpath, mtmp, oldl1, remt, outname,
273 mem, levels2, flushed, manifest_token, frozen, batch_cap, mem_count, frozen_count)
274
275def compact_goA.written(
276 result: Result<&1, &1, U32 & String, Unit>,
277 +dir: String,
278 +l1dir: String,
279 +tmp: String,
280 +final: String,
281 +mpath: String,
282 +mtmp: String,
283 +oldl1: List<&2, Sstable.Table>,
284 +remt: List<&2, Sstable.Table>,
285 +outname: String,
286 +mem: MemTable.MemTable,
287 +levels2: List<&2, List<&2, Sstable.Table>>,
288 +flushed: Nat,
289 +manifest_token: String,
290 +frozen: MemTable.MemTable,
291 +batch_cap: Nat,
292 +mem_count: Nat,
293 +frozen_count: Nat
294) -> IO(Result<&1, &1, U32 & String, Db.Db>):
295 match result:
296 case Fail{error}:
297 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
298 case Done{Unit{}}:
299 do IO<Result<&1, &1, U32 & String, Db.Db>>:
300 _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("compact.output_synced"))
301 renamed : Result<&1, &1, U32 & String, Unit> <- Fs.rename(tmp, final)
302 compact_goA.renamed(renamed, dir, l1dir, mpath, mtmp, oldl1, remt,
303 outname, mem, levels2, flushed, manifest_token, frozen, batch_cap, mem_count, frozen_count)
304
305def compact_goA.ensured(
306 result: Result<&1, &1, U32 & String, Unit>,
307 +dir: String,
308 +l1dir: String,
309 +tmp: String,
310 +final: String,
311 +mpath: String,
312 +mtmp: String,
313 +entries: List<&2, MemTable.Entry>,
314 +level: U32,
315 +outname: String,
316 +mem: MemTable.MemTable,
317 +levels2: List<&2, List<&2, Sstable.Table>>,
318 +oldl1: List<&2, Sstable.Table>,
319 +remt: List<&2, Sstable.Table>,
320 +flushed: Nat,
321 +manifest_token: String,
322 +frozen: MemTable.MemTable,
323 +batch_cap: Nat,
324 +mem_count: Nat,
325 +frozen_count: Nat
326) -> IO(Result<&1, &1, U32 & String, Db.Db>):
327 match result:
328 case Fail{error}:
329 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})
330 case Done{Unit{}}:
331 do IO<Result<&1, &1, U32 & String, Db.Db>>:
332 written : Result<&1, &1, U32 & String, Unit> <- SstStreamIo.write_table(tmp, entries, level)
333 compact_goA.written(written, dir, l1dir, tmp, final, mpath, mtmp,
334 oldl1, remt, outname, mem, levels2, flushed, manifest_token, frozen,
335 batch_cap, mem_count, frozen_count)
336
337# Publishes the compacted run and removes obsolete files.
338def compact_goA(
339 +dir: String,
340 +l1dir: String,
341 +tmp: String,
342 +final: String,
343 +mpath: String,
344 +mtmp: String,
345 +entries: List<&2, MemTable.Entry>,
346 +level: U32,
347 +outname: String,
348 +mem: MemTable.MemTable,
349 +levels2: List<&2, List<&2, Sstable.Table>>,
350 +oldl1: List<&2, Sstable.Table>,
351 +remt: List<&2, Sstable.Table>,
352 +flushed: Nat,
353 +manifest_token: String,
354 +frozen: MemTable.MemTable,
355 +batch_cap: Nat,
356 +mem_count: Nat,
357 +frozen_count: Nat,
358) -> IO(Result<&1, &1, U32 & String, Db.Db>):
359 do IO<Result<&1, &1, U32 & String, Db.Db>>:
360 ensured : Result<&1, &1, U32 & String, Unit> <- Flush.ensure_dir(l1dir)
361 compact_goA.ensured(ensured, dir, l1dir, tmp, final, mpath, mtmp,
362 entries, level, outname, mem, levels2, oldl1, remt, flushed,
363 manifest_token, frozen, batch_cap, mem_count, frozen_count)
364
365# Compact pre for the compaction filesystem effects.
366def compact_pre(
367 +dir: String,
368 +mem: MemTable.MemTable,
369 +levels: List<&2, List<&2, Sstable.Table>>,
370 +flushed: Nat,
371 +manifest_token: String,
372 +frozen: MemTable.MemTable,
373 +batch_cap: Nat,
374 +mem_count: Nat,
375 +frozen_count: Nat,
376) -> IO(Result<&1, &1, U32 & String, Db.Db>):
377 +levels2 = Compact.compact_levels(levels)
378 +newl1 = Compact.l1list(levels2)
379 +out = Compact.hd_tbl(newl1)
380 +remt = Compact.tail_of(newl1)
381 +l1dir = dir ++ "/l1"
382 +outname = "l1-" ++ Nat.show(flushed) ++ ".tbl"
383 +tmp = l1dir ++ "/" ++ outname ++ ".tmp"
384 +final = l1dir ++ "/" ++ outname
385 compact_goA(
386 dir, l1dir, tmp, final, dir ++ "/MANIFEST", dir ++ "/MANIFEST.tmp",
387 Db.table_entries(out), 1, outname, mem, levels2, Compact.l1list(levels), remt, flushed,
388 manifest_token, frozen, batch_cap, mem_count, frozen_count,
389 )
390
391# Compact trig for the compaction filesystem effects.
392def compact_trig(
393 gt: Bool,
394 +dir: String,
395 +mem: MemTable.MemTable,
396 +levels: List<&2, List<&2, Sstable.Table>>,
397 +flushed: Nat,
398 +manifest_token: String,
399 +frozen: MemTable.MemTable,
400 +batch_cap: Nat,
401 +bcache: List<&2, Db.BEntry>,
402 +mem_count: Nat,
403 +frozen_count: Nat,
404) -> IO(Result<&1, &1, U32 & String, Db.Db>):
405 match gt:
406 case True{}:
407 compact_pre(dir, mem, levels, flushed, manifest_token, frozen, batch_cap, mem_count, frozen_count)
408 case False{}:
409 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Done{Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}})
410
411# Handle compact in the compaction filesystem effects.
412def compact(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
413 match db:
414 case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
415 match levels:
416 case Nil{}:
417 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Done{Db.Db{dir, mem, frozen, batch_cap, bcache, Nil{}, flushed, manifest_token, mem_count, frozen_count}})
418 case Con{+l0, r1}:
419 compact_trig(Nat.is_lt(4n, List.length(&2, Sstable.Table, l0)), dir, mem, Con{l0, r1}, flushed, manifest_token, frozen, batch_cap, bcache, mem_count, frozen_count)