src/CompactIo.bend
mylsm-lsm-store@0.4.0.0 · 160 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.4.0.0/src/Compact.bend as Compact
3import mylsm-lsm-store@0.4.0.0/src/MemTable.bend as MemTable
4import mylsm-lsm-store@0.4.0.0/src/Sstable.bend as Sstable
5import mylsm-lsm-store@0.4.0.0/src/SstStreamIo.bend as SstStreamIo
6import mylsm-lsm-store@0.4.0.0/src/Wal.bend as Wal
7import mylsm-lsm-store@0.4.0.0/src/Manifest.bend as Manifest
8import mylsm-lsm-store@0.4.0.0/src/Flush.bend as Flush
9import mylsm-lsm-store@0.4.0.0/src/Fs.bend as Fs
10import mylsm-lsm-store@0.4.0.0/src/Db.bend as Db
11import mylsm-lsm-store@0.4.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# Compact goB for the compaction filesystem effects.
24def compact_goB(
25 +dir: String,
26 +mpath: String,
27 +mtmp: String,
28 +m2: Manifest.Manifest,
29 +dels: List<&2, String>,
30 +mem: MemTable.MemTable,
31 +levels2: List<&2, List<&2, Sstable.Table>>,
32 +flushed: Nat,
33 +frozen: MemTable.MemTable,
34 +batch_cap: Nat,
35 +mem_count: Nat,
36 +frozen_count: Nat,
37) -> IO(Result<&1, &1, U32 & String, Db.Db>):
38 do IO<Result<&1, &1, U32 & String, Db.Db>>:
39 _v1 : Unit <- IO.try(Unit, Flush.write_manifest(mtmp, m2))
40 _cp1 : Unit <- IO.try(Unit, CrashPoint.hit("compact.manifest_synced"))
41 _v2 : Unit <- IO.try(Unit, Fs.rename(mtmp, mpath))
42 _v3 : Unit <- IO.try(Unit, Fs.fsync(dir))
43 _cp2 : Unit <- IO.try(Unit, CrashPoint.hit("compact.manifest_published"))
44 _ign : Result<&1, &1, U32 & String, Unit> <- remove_list(dels)
45 return Done{Db.Db{dir, mem, frozen, batch_cap, Nil{}, levels2, Nat.add(flushed, 1n), Manifest.token(m2), mem_count, frozen_count}}
46
47# Handle drift go in the compaction filesystem effects.
48def drift_go(
49 ok: Bool,
50 +dir: String,
51 +mpath: String,
52 +mtmp: String,
53 +mfst: Manifest.Manifest,
54 +oldl1: List<&2, Sstable.Table>,
55 +remt: List<&2, Sstable.Table>,
56 +outname: String,
57 +mem: MemTable.MemTable,
58 +levels2: List<&2, List<&2, Sstable.Table>>,
59 +flushed: Nat,
60 +frozen: MemTable.MemTable,
61 +batch_cap: Nat,
62 +mem_count: Nat,
63 +frozen_count: Nat,
64) -> IO(Result<&1, &1, U32 & String, Db.Db>):
65 match ok:
66 case True{}:
67 +new_mfst = Compact.mfst_new(mfst, Compact.rem_names(oldl1, Compact.ml1(mfst), remt), outname)
68 +removed = Compact.abs_names(oldl1, Compact.ml1(mfst), remt)
69 +dels = List.append(&2, String, Compact.prefix(Compact.ml0(mfst), dir ++ "/l0"), Compact.prefix(removed, dir ++ "/l1"))
70 compact_goB(dir, mpath, mtmp, new_mfst, dels, mem, levels2, flushed, frozen, batch_cap, mem_count, frozen_count)
71 case False{}:
72 IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(2n), "manifest drift")})
73
74# Compact goA for the compaction filesystem effects.
75def compact_goA(
76 +dir: String,
77 +l1dir: String,
78 +tmp: String,
79 +final: String,
80 +mpath: String,
81 +mtmp: String,
82 +entries: List<&2, MemTable.Entry>,
83 +level: U32,
84 +outname: String,
85 +mem: MemTable.MemTable,
86 +levels2: List<&2, List<&2, Sstable.Table>>,
87 +oldl1: List<&2, Sstable.Table>,
88 +remt: List<&2, Sstable.Table>,
89 +flushed: Nat,
90 +manifest_token: String,
91 +frozen: MemTable.MemTable,
92 +batch_cap: Nat,
93 +mem_count: Nat,
94 +frozen_count: Nat,
95) -> IO(Result<&1, &1, U32 & String, Db.Db>):
96 do IO<Result<&1, &1, U32 & String, Db.Db>>:
97 _u1 : Unit <- IO.try(Unit, Flush.ensure_dir(l1dir))
98 _u2 : Unit <- IO.try(Unit, SstStreamIo.write_table(tmp, entries, level))
99 _cp1 : Unit <- IO.try(Unit, CrashPoint.hit("compact.output_synced"))
100 _u3 : Unit <- IO.try(Unit, Fs.rename(tmp, final))
101 _u4 : Unit <- IO.try(Unit, Fs.fsync(l1dir))
102 _cp2 : Unit <- IO.try(Unit, CrashPoint.hit("compact.output_published"))
103 +m : Manifest.Manifest <- IO.try(Manifest.Manifest, Flush.load_mfst(mpath))
104 drift_go(String.eq(Manifest.token(m), manifest_token), dir, mpath, mtmp, m, oldl1, remt, outname, mem, levels2, flushed, frozen, batch_cap, mem_count, frozen_count)
105
106# Compact pre for the compaction filesystem effects.
107def compact_pre(
108 +dir: String,
109 +mem: MemTable.MemTable,
110 +levels: List<&2, List<&2, Sstable.Table>>,
111 +flushed: Nat,
112 +manifest_token: String,
113 +frozen: MemTable.MemTable,
114 +batch_cap: Nat,
115 +mem_count: Nat,
116 +frozen_count: Nat,
117) -> IO(Result<&1, &1, U32 & String, Db.Db>):
118 +levels2 = Compact.compact_levels(levels)
119 +newl1 = Compact.l1list(levels2)
120 +out = Compact.hd_tbl(newl1)
121 +remt = Compact.tail_of(newl1)
122 +l1dir = dir ++ "/l1"
123 +outname = "l1-" ++ Nat.show(flushed) ++ ".tbl"
124 +tmp = l1dir ++ "/" ++ outname ++ ".tmp"
125 +final = l1dir ++ "/" ++ outname
126 compact_goA(
127 dir, l1dir, tmp, final, dir ++ "/MANIFEST", dir ++ "/MANIFEST.tmp",
128 Db.table_entries(out), 1, outname, mem, levels2, Compact.l1list(levels), remt, flushed,
129 manifest_token, frozen, batch_cap, mem_count, frozen_count,
130 )
131
132# Compact trig for the compaction filesystem effects.
133def compact_trig(
134 gt: Bool,
135 +dir: String,
136 +mem: MemTable.MemTable,
137 +levels: List<&2, List<&2, Sstable.Table>>,
138 +flushed: Nat,
139 +manifest_token: String,
140 +frozen: MemTable.MemTable,
141 +batch_cap: Nat,
142 +bcache: List<&2, Db.BEntry>,
143 +mem_count: Nat,
144 +frozen_count: Nat,
145) -> IO(Result<&1, &1, U32 & String, Db.Db>):
146 match gt:
147 case True{}:
148 compact_pre(dir, mem, levels, flushed, manifest_token, frozen, batch_cap, mem_count, frozen_count)
149 case False{}:
150 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}})
151
152# Handle compact in the compaction filesystem effects.
153def compact(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):
154 match db:
155 case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:
156 match levels:
157 case Nil{}:
158 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}})
159 case Con{+l0, r1}:
160 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)