src/legacy/ManifestIo.bend

mylsm-lsm-store@0.6.0.1 · 278 lines · 25 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/Manifest.bend as Manifest
3import mylsm-lsm-store@0.6.0.1/src/StorageBytes.bend as StorageBytes
4import mylsm-lsm-store@0.6.0.1/src/Fs.bend as Fs
5import mylsm-lsm-store@0.6.0.1/src/CrashPoint.bend as CrashPoint
6import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7
8# Represent ManifestReadState data used by the memtable flush.
9type ManifestReadState is Type:
10  ManifestNeed{file: File, remaining: Nat, chunks: List<&1, Bytes.Bytes>}
11  ManifestGot{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>, remaining: Nat, chunks: List<&1, Bytes.Bytes>}
12
13def write_tail.chmod(
14  result: Result<&1, &1, U32 & String, Unit>,
15  +path: String
16) -> IO(Result<&1, &1, U32 & String, Unit>):
17  match result:
18    case Fail{error}:
19      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
20    case Done{Unit{}}:
21      Fs.chmod(path, U32.from_nat(384n))
22
23def write_tail.fsync(
24  result: Result<&1, &1, U32 & String, Unit>,
25  +path: String
26) -> IO(Result<&1, &1, U32 & String, Unit>):
27  match result:
28    case Fail{error}:
29      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
30    case Done{Unit{}}:
31      do IO<Result<&1, &1, U32 & String, Unit>>:
32        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path)
33        write_tail.chmod(synced, path)
34
35# Writes the remaining bytes of a file buffer.
36def write_tail(
37  fr: File & Result<&1, &1, U32 & String, Unit>,
38  +path: String
39) -> IO(Result<&1, &1, U32 & String, Unit>):
40  match fr:
41    case (file, result):
42      do IO<Result<&1, &1, U32 & String, Unit>>:
43        _closed : Unit <- File.close(file)
44        write_tail.fsync(result, path)
45
46def write_file.opened(
47  opened: Result<&1, &1, U32 & String, File>,
48  +path: String,
49  +data: String
50) -> IO(Result<&1, &1, U32 & String, Unit>):
51  match opened:
52    case Fail{error}:
53      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
54    case Done{file}:
55      do IO<Result<&1, &1, U32 & String, Unit>>:
56        written : File & Result<&1, &1, U32 & String, Unit> <- File.write(file, data)
57        write_tail(written, path)
58
59# Creates and syncs one storage file.
60def write_file(+path: String, +data: String) -> IO(Result<&1, &1, U32 & String, Unit>):
61  do IO<Result<&1, &1, U32 & String, Unit>>:
62    opened : Result<&1, &1, U32 & String, File> <- File.open(path, "w")
63    write_file.opened(opened, path, data)
64
65def write_manifest.bytes.opened(
66  opened: Result<&1, &1, U32 & String, File>,
67  +path: String,
68  bytes: Bytes.Bytes
69) -> IO(Result<&1, &1, U32 & String, Unit>):
70  match opened:
71    case Fail{error}:
72      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
73    case Done{file}:
74      do IO<Result<&1, &1, U32 & String, Unit>>:
75        written : File & Result<&1, &1, U32 & String, Unit> <- Fs.write_bytes(file, bytes)
76        write_tail(written, path)
77
78def write_manifest.bytes(+path: String, bytes: Bytes.Bytes) -> IO(Result<&1, &1, U32 & String, Unit>):
79  do IO<Result<&1, &1, U32 & String, Unit>>:
80    opened : Result<&1, &1, U32 & String, File> <- File.open(path, "w")
81    write_manifest.bytes.opened(opened, path, bytes)
82
83def write_manifest.encoded(
84  path: String,
85  encoded: Result<&1, &1, Manifest.Error, Bytes.Bytes>
86) -> IO(Result<&1, &1, U32 & String, Unit>):
87  match encoded:
88    case Fail{_}:
89      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{(U32.from_nat(1n), "cannot encode manifest")})
90    case Done{bytes}:
91      write_manifest.bytes(path, bytes)
92
93# Write manifest for the memtable flush.
94def write_manifest(path: String, manifest: Manifest.Manifest) -> IO(Result<&1, &1, U32 & String, Unit>):
95  write_manifest.encoded(path, Manifest.serialize(manifest))
96
97# Read tail3 for the memtable flush.
98def read_tail3(fr: (File & Result<&1, &1, U32 & String, String>)) -> IO(Result<&1, &1, U32 & String, String>):
99  match fr:
100    case (f3, r1):
101      do IO<Result<&1, &1, U32 & String, String>>:
102        _cls : Unit <- File.close(f3)
103        IO.pure(Result<&1, &1, U32 & String, String>, r1)
104
105
106# Base File.read decodes each returned chunk independently. A chunk boundary can
107# split a multi-byte UTF-8 character, so table files are read in one bounded call
108# Reads a bounded storage file and returns its bytes.
109def read_file.opened(
110  opened: Result<&1, &1, U32 & String, File>
111) -> IO(Result<&1, &1, U32 & String, String>):
112  match opened:
113    case Fail{error}:
114      IO.pure(Result<&1, &1, U32 & String, String>, Fail{error})
115    case Done{file}:
116      do IO<Result<&1, &1, U32 & String, String>>:
117        result : File & Result<&1, &1, U32 & String, String> <- File.read(file, U32.from_nat(4294967295n))
118        read_tail3(result)
119
120# Reads a bounded storage file and returns its bytes.
121def read_file(+path: String) -> IO(Result<&1, &1, U32 & String, String>):
122  do IO<Result<&1, &1, U32 & String, String>>:
123    opened : Result<&1, &1, U32 & String, File> <- File.open(path, "r")
124    read_file.opened(opened)
125
126# Classifies the Manifest parse result before recovery.
127def mfst_parsed(parsed: Manifest.ParseResult) -> Result<&1, &1, U32 & String, Manifest.Manifest>:
128  match parsed:
129    case Manifest.Parsed{manifest}:
130      Done{manifest}
131    case Manifest.UnsupportedVersion{_}:
132      Fail{(Manifest.unsupported_version_error_code(), "unsupported manifest version")}
133    case Manifest.Malformed{}:
134      Fail{(Manifest.malformed_error_code(), "malformed manifest")}
135    case Manifest.ParseLimitExceeded{}:
136      Fail{(Manifest.parse_limit_error_code(), "manifest size limit exceeded")}
137
138def manifest.read.result(
139  pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
140) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
141  match pair:
142    case (file, Fail{error}):
143      do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
144        _closed : Unit <- File.close(file)
145        return Fail{error}
146    case (file, Done{bytes}):
147      do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
148        _closed : Unit <- File.close(file)
149        return mfst_parsed(Manifest.parse_result(bytes))
150
151def manifest.read.loop(fuel: Nat, state: ManifestReadState) -> IO(File & Result<&1, &1, U32 & String, Bytes.Bytes>):
152  match fuel:
153    case 0n:
154      match state:
155        case ManifestNeed{file, +remaining, chunks}:
156          match remaining:
157            case 0n:
158              IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Done{Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks))}))
159            case 1n+_:
160              IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{(U32.from_nat(7n), "bounded manifest read exhausted")}))
161        case ManifestGot{pair, _, _}:
162          match pair:
163            case (file, Fail{error}):
164              IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{error}))
165            case (file, Done{_}):
166              IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{(U32.from_nat(7n), "bounded manifest read exhausted")}))
167    case 1n+rest:
168      match state:
169        case ManifestNeed{file, +remaining, chunks}:
170          match remaining:
171            case 0n:
172              IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Done{Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks))}))
173            case 1n+_:
174              do IO<File & Result<&1, &1, U32 & String, Bytes.Bytes>>:
175                pair : File & Result<&1, &1, U32 & String, Bytes.Bytes> <- Fs.read_bytes(file, U32.from_nat(remaining))
176                manifest.read.loop(rest, ManifestGot{pair, remaining, chunks})
177        case ManifestGot{pair, +remaining, chunks}:
178          match pair:
179            case (file, Fail{error}):
180              IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{error}))
181            case (file, Done{Bytes.Bytes{0, _}}):
182              IO.pure(File & Result<&1, &1, U32 & String, Bytes.Bytes>, (file, Fail{(U32.from_nat(8n), "truncated manifest file")}))
183            case (file, Done{Bytes.Bytes{+len, buf}}):
184              manifest.read.loop(rest, ManifestNeed{file, Nat.sub(remaining, U32.to_nat(len)), Con{Bytes.Bytes{len, buf}, chunks}})
185
186def manifest.read.size.run(file: File, +size: Nat) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
187  do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
188    pair : File & Result<&1, &1, U32 & String, Bytes.Bytes> <- manifest.read.loop(Nat.add(size, 1n), ManifestNeed{file, size, Nil{}})
189    manifest.read.result(pair)
190
191def manifest.read.size.valid(
192  file: File,
193  size: Nat,
194  within: Bool
195) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
196  match within:
197    case False{}:
198      do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
199        _closed : Unit <- File.close(file)
200        return Fail{(U32.from_nat(9n), "manifest too large")}
201    case True{}:
202      manifest.read.size.run(file, size)
203
204def manifest.read.size(
205  file: File,
206  result: Result<&1, &1, U32 & String, Nat>
207) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
208  match result:
209    case Fail{error}:
210      do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
211        _closed : Unit <- File.close(file)
212        return Fail{error}
213    case Done{+size}:
214      manifest.read.size.valid(file, size, Nat.is_le(size, U32.to_nat(StorageBytes.MAX_MANIFEST_BYTES())))
215
216# Handle mfst open error in the memtable flush.
217def mfst_open_error(
218  missing: Bool,
219  +code: U32
220) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
221  match missing:
222    case True{}:
223      IO.pure(Result<&1, &1, U32 & String, Manifest.Manifest>, Done{Manifest.M{Nil{}}})
224    case False{}:
225      IO.pure(Result<&1, &1, U32 & String, Manifest.Manifest>, Fail{(code, "cannot open manifest")})
226
227# Handle mfst opened in the memtable flush.
228def mfst_opened(
229  path: String,
230  result: Result<&1, &1, U32 & String, File>
231) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
232  match result:
233    case Fail{(+code, _)}:
234      mfst_open_error(U32.is_eq(code, 2), code)
235    case Done{file}:
236      do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
237        size : Result<&1, &1, U32 & String, Nat> <- Fs.file_size(path)
238        manifest.read.size(file, size)
239
240# Handle load mfst in the memtable flush.
241def load_mfst(+path: String) -> IO(Result<&1, &1, U32 & String, Manifest.Manifest>):
242  do IO<Result<&1, &1, U32 & String, Manifest.Manifest>>:
243    opened : Result<&1, &1, U32 & String, File> <- File.open(path, "r")
244    mfst_opened(path, opened)
245
246# Handle dir tail in the memtable flush.
247def dir_tail(res: Result<&1, &1, U32 & String, Nat>, +path: String) -> IO(Result<&1, &1, U32 & String, Unit>):
248  match res:
249    case Done{n}:
250      IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
251    case Fail{e}:
252      Fs.make_dir(path)
253
254# Handle ensure dir in the memtable flush.
255def ensure_dir(+path: String) -> IO(Result<&1, &1, U32 & String, Unit>):
256  do IO<Result<&1, &1, U32 & String, Unit>>:
257    r : Result<&1, &1, U32 & String, Nat> <- Fs.read_dir_count(path)
258    dir_tail(r, path)
259
260# Handle ldir of in the database recovery effects.
261def ldir_of(idx: Nat) -> Maybe<&2, String>:
262  match idx:
263    case 0n:
264      Some{"l0"}
265    case 1n+m:
266      match m:
267        case 0n:
268          Some{"l1"}
269        case 1n+p:
270          match p:
271            case 0n:
272              Some{"l2"}
273            case 1n+q:
274              match q:
275                case 0n:
276                  Some{"l3"}
277                case 1n+r:
278                  None{}