src/SstStreamIo.bend
mylsm-lsm-store@0.4.0.0 · 491 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.4.0.0/src/Fs.bend as Fs
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/SstFile.bend as SstFile
6import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7
8# Map pure format errors to the storage IO error domain.
9def codec_error(error: SstFile.Error) -> U32 & String:
10 match error:
11 case SstFile.InvalidLevel{}:
12 (U32.from_nat(3n), "invalid SST level")
13 case SstFile.InvalidOrder{}:
14 (U32.from_nat(3n), "SST keys are not strictly ordered")
15 case SstFile.TooLarge{}:
16 (U32.from_nat(3n), "SST exceeds a size limit")
17 case SstFile.Malformed{}:
18 (U32.from_nat(3n), "malformed v3 SST")
19 case SstFile.InvalidUtf8{}:
20 (U32.from_nat(3n), "invalid UTF-8 in v3 SST")
21 case SstFile.ChecksumMismatch{}:
22 (U32.from_nat(3n), "v3 SST checksum mismatch")
23
24# Represent ExactResult data used by the streaming SSTable filesystem effects.
25type ExactResult is Type:
26 ExactResult{bytes: Bytes.Bytes, complete: Bool}
27
28# Represent ExactState data used by the streaming SSTable filesystem effects.
29type ExactState is Type:
30 ExactNeed{file: File, remaining: Nat, chunks: List<&1, Bytes.Bytes>}
31 ExactRead{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
32 remaining: Nat, chunks: List<&1, Bytes.Bytes>}
33
34def exact.complete(
35 file: File,
36 chunks: List<&1, Bytes.Bytes>,
37 complete: Bool
38) -> IO(File & Result<&1, &1, U32 & String, ExactResult>):
39 IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
40 (file, Done{ExactResult{Bytes.concat(List.reverse(&1, Bytes.Bytes, chunks)), complete}}))
41
42def exact.loop(
43 fuel: Nat,
44 state: ExactState
45) -> IO(File & Result<&1, &1, U32 & String, ExactResult>):
46 match fuel:
47 case 0n:
48 match state:
49 case ExactNeed{file, remaining, chunks}:
50 match remaining:
51 case 0n:
52 exact.complete(file, chunks, True{})
53 case 1n+_:
54 IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
55 (file, Fail{(U32.from_nat(7n), "bounded read exhausted")}))
56 case ExactRead{pair, _, _}:
57 match pair:
58 case (file, Fail{error}):
59 IO.pure(File & Result<&1, &1, U32 & String, ExactResult>, (file, Fail{error}))
60 case (file, Done{_}):
61 IO.pure(File & Result<&1, &1, U32 & String, ExactResult>,
62 (file, Fail{(U32.from_nat(7n), "bounded read exhausted")}))
63 case 1n+rest:
64 match state:
65 case ExactNeed{file, +remaining, chunks}:
66 match remaining:
67 case 0n:
68 exact.complete(file, chunks, True{})
69 case 1n+_:
70 do IO<File & Result<&1, &1, U32 & String, ExactResult>>:
71 next : File & Result<&1, &1, U32 & String, Bytes.Bytes> <-
72 Fs.read_bytes(file, U32.from_nat(Nat.min(1048576n, remaining)))
73 exact.loop(rest, ExactRead{next, remaining, chunks})
74 case ExactRead{pair, remaining, chunks}:
75 match pair:
76 case (file, Fail{error}):
77 IO.pure(File & Result<&1, &1, U32 & String, ExactResult>, (file, Fail{error}))
78 case (file, Done{Bytes.Bytes{0, _}}):
79 exact.complete(file, chunks, False{})
80 case (file, Done{Bytes.Bytes{+len, buf}}):
81 exact.loop(rest, ExactNeed{file, (remaining - U32.to_nat(len) : Nat),
82 Con{Bytes.Bytes{len, buf}, chunks}})
83
84def exact.start(file: File, +needed: Nat) -> IO(File & Result<&1, &1, U32 & String, ExactResult>):
85 exact.loop(Nat.add(Nat.mul(2n, needed), 1n), ExactNeed{file, needed, Nil{}})
86
87# Represent StreamReadState data used by the streaming SSTable filesystem effects.
88type StreamReadState is Type:
89 ReadHeader{file: File, size: Nat}
90 ReadHeaderResult{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>, size: Nat}
91 ReadBlocks{file: File, size: Nat, used: Nat, parser: SstFile.StreamState}
92 ReadBlockHeaderResult{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
93 size: Nat, used: Nat, parser: SstFile.StreamState}
94 ReadBlockBody{file: File, size: Nat, used: Nat,
95 parser: SstFile.StreamState, header: Bytes.Bytes, tail_len: U32}
96 ReadBlockBodyResult{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
97 size: Nat, used: Nat, parser: SstFile.StreamState,
98 header: Bytes.Bytes, tail_len: U32}
99 ReadFooterResult{pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
100 size: Nat, used: Nat, parser: SstFile.StreamState}
101 ReadClose{file: File, result: Result<&1, &1, U32 & String, Sstable.Table>}
102
103def read.exact.complete(bytes: Bytes.Bytes, complete: Bool) -> Result<&1, &1, U32 & String, Bytes.Bytes>:
104 match complete:
105 case True{}:
106 Done{bytes}
107 case False{}:
108 Fail{(U32.from_nat(3n), "truncated SST")}
109
110def read.exact.state(
111 pair: File & Result<&1, &1, U32 & String, ExactResult>
112) -> File & Result<&1, &1, U32 & String, Bytes.Bytes>:
113 match pair:
114 case (file, Fail{error}):
115 (file, Fail{error})
116 case (file, Done{ExactResult{bytes, complete}}):
117 (file, read.exact.complete(bytes, complete))
118
119def read.header.parsed(
120 file: File,
121 +size: Nat,
122 result: Result<&1, &1, SstFile.Error, SstFile.StreamState>
123) -> StreamReadState:
124 match result:
125 case Fail{error}:
126 ReadClose{file, Fail{codec_error(error)}}
127 case Done{parser}:
128 ReadBlocks{file, size, 20n, parser}
129
130def read.header.state(
131 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
132 +size: Nat
133) -> StreamReadState:
134 match pair:
135 case (file, Fail{error}):
136 ReadClose{file, Fail{error}}
137 case (file, Done{bytes}):
138 read.header.parsed(file, size, SstFile.stream.header(bytes))
139
140def read.block.header.bound(
141 file: File,
142 +size: Nat,
143 +used: Nat,
144 parser: SstFile.StreamState,
145 header: Bytes.Bytes,
146 +tail_len: U32,
147 valid: Bool
148) -> StreamReadState:
149 match valid:
150 case False{}:
151 ReadClose{file, Fail{(U32.from_nat(3n), "truncated SST block")}}
152 case True{}:
153 ReadBlockBody{file, size, used, parser, header, tail_len}
154
155def read.block.header.size(
156 file: File,
157 +size: Nat,
158 +used: Nat,
159 parser: SstFile.StreamState,
160 header: Bytes.Bytes,
161 tail: Maybe<&2, U32>
162) -> StreamReadState:
163 match tail:
164 case None{}:
165 ReadClose{file, Fail{(U32.from_nat(3n), "invalid SST block size")}}
166 case Some{+tail_len}:
167 read.block.header.bound(file, size, used, parser, header, tail_len,
168 Nat.is_le(Nat.add(Nat.add(used, 8n), U32.to_nat(tail_len)),
169 Nat.sub(size, 40n)))
170
171def read.block.header.tail(
172 saved: Bytes.Bytes,
173 tail: Maybe<&2, U32>
174) -> Bytes.Bytes & Maybe<&2, U32>:
175 (saved, tail)
176
177def read.block.header.inspected(
178 pair: Bytes.Bytes & Bytes.Bytes
179) -> Bytes.Bytes & Maybe<&2, U32>:
180 match pair:
181 case (saved, inspect):
182 read.block.header.tail(saved, SstFile.stream.block.tail_size(inspect))
183
184def read.block.header.copied(
185 file: File,
186 +size: Nat,
187 +used: Nat,
188 parser: SstFile.StreamState,
189 inspected: Bytes.Bytes & Maybe<&2, U32>
190) -> StreamReadState:
191 match inspected:
192 case (saved, None{}):
193 ReadClose{file, Fail{(U32.from_nat(3n), "invalid SST block size")}}
194 case (saved, Some{+tail_len}):
195 read.block.header.size(file, size, used, parser, saved,
196 Some{tail_len})
197
198def read.block.header.copy(
199 file: File,
200 +size: Nat,
201 +used: Nat,
202 parser: SstFile.StreamState,
203 header: Bytes.Bytes
204) -> StreamReadState:
205 read.block.header.copied(file, size, used, parser,
206 read.block.header.inspected(SstFile.bytes.clone.split(header)))
207
208def read.block.header.state(
209 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
210 +size: Nat,
211 +used: Nat,
212 parser: SstFile.StreamState
213) -> StreamReadState:
214 match pair:
215 case (file, Fail{error}):
216 ReadClose{file, Fail{error}}
217 case (file, Done{header}):
218 read.block.header.copy(file, size, used, parser, header)
219
220def read.block.body.parsed(
221 file: File,
222 +size: Nat,
223 +used: Nat,
224 +tail_len: U32,
225 result: Result<&1, &1, SstFile.Error, SstFile.StreamState>
226) -> StreamReadState:
227 match result:
228 case Fail{error}:
229 ReadClose{file, Fail{codec_error(error)}}
230 case Done{parser}:
231 ReadBlocks{file, size, Nat.add(used,
232 Nat.add(8n, U32.to_nat(tail_len))), parser}
233
234def read.block.body.state(
235 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
236 +size: Nat,
237 +used: Nat,
238 parser: SstFile.StreamState,
239 header: Bytes.Bytes,
240 +tail_len: U32
241) -> StreamReadState:
242 match pair:
243 case (file, Fail{error}):
244 ReadClose{file, Fail{error}}
245 case (file, Done{body}):
246 read.block.body.parsed(file, size, used, tail_len,
247 SstFile.stream.block(parser, Bytes.concat([header, body])))
248
249def read.footer.parsed(
250 file: File,
251 result: Result<&1, &1, SstFile.Error, Sstable.Table>
252) -> StreamReadState:
253 match result:
254 case Fail{error}:
255 ReadClose{file, Fail{codec_error(error)}}
256 case Done{table}:
257 ReadClose{file, Done{table}}
258
259def read.footer.size(
260 file: File,
261 parser: SstFile.StreamState,
262 bytes: Bytes.Bytes,
263 valid: Bool
264) -> StreamReadState:
265 match valid:
266 case False{}:
267 ReadClose{file, Fail{(U32.from_nat(3n), "malformed SST size")}}
268 case True{}:
269 read.footer.parsed(file, SstFile.stream.footer(parser, bytes))
270
271def read.footer.state(
272 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>,
273 +size: Nat,
274 +used: Nat,
275 parser: SstFile.StreamState
276) -> StreamReadState:
277 match pair:
278 case (file, Fail{error}):
279 ReadClose{file, Fail{error}}
280 case (file, Done{bytes}):
281 read.footer.size(file, parser, bytes,
282 Nat.is_eq(Nat.add(used, 40n), size))
283
284def read.exhausted.file(file: File) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
285 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
286 _closed : Unit <- File.close(file)
287 return Fail{(U32.from_nat(7n), "bounded SST read exhausted")}
288
289def read.exhausted.pair(
290 pair: File & Result<&1, &1, U32 & String, Bytes.Bytes>
291) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
292 match pair:
293 case (file, _):
294 read.exhausted.file(file)
295
296def read.exhausted(state: StreamReadState) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
297 match state:
298 case ReadHeader{file, _}:
299 read.exhausted.file(file)
300 case ReadHeaderResult{pair, _}:
301 read.exhausted.pair(pair)
302 case ReadBlocks{file, _, _, _}:
303 read.exhausted.file(file)
304 case ReadBlockHeaderResult{pair, _, _, _}:
305 read.exhausted.pair(pair)
306 case ReadBlockBody{file, _, _, _, _, _}:
307 read.exhausted.file(file)
308 case ReadBlockBodyResult{pair, _, _, _, _, _}:
309 read.exhausted.pair(pair)
310 case ReadFooterResult{pair, _, _, _}:
311 read.exhausted.pair(pair)
312 case ReadClose{file, result}:
313 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
314 _closed : Unit <- File.close(file)
315 return result
316
317def read.loop(fuel: Nat, state: StreamReadState) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
318 match fuel:
319 case 0n:
320 read.exhausted(state)
321 case 1n+rest:
322 match state:
323 case ReadHeader{file, size}:
324 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
325 pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, 20n)
326 read.loop(rest, read.header.state(read.exact.state(pair), size))
327 case ReadHeaderResult{pair, size}:
328 read.loop(rest, read.header.state(pair, size))
329 case ReadBlocks{file, size, used, parser}:
330 match parser:
331 case SstFile.StreamState{_, _, _, 0, _, _, _, _}:
332 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
333 pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, 40n)
334 read.loop(rest, read.footer.state(read.exact.state(pair),
335 size, used, parser))
336 case SstFile.StreamState{+level, +total, +block_count, +remaining,
337 +entries_left, +previous, entries, hashes}:
338 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
339 pair : File & Result<&1, &1, U32 & String, ExactResult> <- exact.start(file, 8n)
340 read.loop(rest, read.block.header.state(read.exact.state(pair),
341 size, used, SstFile.StreamState{level, total, block_count,
342 remaining, entries_left, previous, entries, hashes}))
343 case ReadBlockHeaderResult{pair, size, used, parser}:
344 read.loop(rest, read.block.header.state(pair, size, used, parser))
345 case ReadBlockBody{file, size, used, parser, header, +tail_len}:
346 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
347 pair : File & Result<&1, &1, U32 & String, ExactResult> <-
348 exact.start(file, U32.to_nat(tail_len))
349 read.loop(rest, ReadBlockBodyResult{read.exact.state(pair),
350 size, used, parser, header, tail_len})
351 case ReadBlockBodyResult{pair, size, used, parser, header, tail_len}:
352 read.loop(rest, read.block.body.state(pair, size, used, parser,
353 header, tail_len))
354 case ReadFooterResult{pair, size, used, parser}:
355 read.loop(rest, read.footer.state(pair, size, used, parser))
356 case ReadClose{file, result}:
357 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
358 _closed : Unit <- File.close(file)
359 return result
360
361def read.table.size.checked(
362 +path: String,
363 +size: Nat,
364 valid: Bool
365) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
366 match valid:
367 case False{}:
368 IO.pure(Result<&1, &1, U32 & String, Sstable.Table>,
369 Fail{(U32.from_nat(3n), "malformed or oversized SST")})
370 case True{}:
371 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
372 file : File <- IO.try(File, File.open(path, "r"))
373 read.loop(Nat.add(Nat.mul(2n, size), 16n), ReadHeader{file, size})
374
375def read.table.size(+path: String, +size: Nat) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
376 read.table.size.checked(path, size,
377 Nat.is_le(60n, size) && Nat.is_le(size, 4294967295n))
378
379# Read table for the streaming SSTable filesystem effects.
380def read_table(+path: String) -> IO(Result<&1, &1, U32 & String, Sstable.Table>):
381 do IO<Result<&1, &1, U32 & String, Sstable.Table>>:
382 size : Nat <- IO.try(Nat, Fs.file_size(path))
383 read.table.size(path, size)
384
385def write.finish.chmod(
386 result: Result<&1, &1, U32 & String, Unit>
387) -> IO(Result<&1, &1, U32 & String, Unit>):
388 match result:
389 case Fail{error}:
390 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
391 case Done{Unit{}}:
392 IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})
393
394
395def write.finish.synced(
396 +path: String,
397 result: Result<&1, &1, U32 & String, Unit>
398) -> IO(Result<&1, &1, U32 & String, Unit>):
399 match result:
400 case Fail{error}:
401 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})
402 case Done{Unit{}}:
403 do IO<Result<&1, &1, U32 & String, Unit>>:
404 changed : Result<&1, &1, U32 & String, Unit> <-
405 Fs.chmod(path, U32.from_nat(384n))
406 write.finish.chmod(changed)
407
408def write.finish(
409 file: File,
410 +path: String
411) -> IO(Result<&1, &1, U32 & String, Unit>):
412 do IO<Result<&1, &1, U32 & String, Unit>>:
413 _closed : Unit <- File.close(file)
414 synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(path)
415 write.finish.synced(path, synced)
416
417# Linear writer state keeps the file handle paired with the encoder result.
418type WriteState is Type:
419 WriteNext{file: File, encoder: SstFile.Encoder}
420 WriteOutput{file: File,
421 pair: SstFile.Encoder & Maybe<&1, Bytes.Bytes>}
422 WriteResult{pair: File & Result<&1, &1, U32 & String, Unit>,
423 encoder: SstFile.Encoder}
424
425def write.step(
426 fuel: Nat,
427 +path: String,
428 state: WriteState
429) -> IO(Result<&1, &1, U32 & String, Unit>):
430 match fuel:
431 case 0n:
432 match state:
433 case WriteNext{file, _}:
434 do IO<Result<&1, &1, U32 & String, Unit>>:
435 _closed : Unit <- File.close(file)
436 return Fail{(U32.from_nat(7n), "table exceeds packed write limit")}
437 case WriteOutput{file, _}:
438 do IO<Result<&1, &1, U32 & String, Unit>>:
439 _closed : Unit <- File.close(file)
440 return Fail{(U32.from_nat(7n), "table exceeds packed write limit")}
441 case WriteResult{pair, _}:
442 match pair:
443 case (file, _):
444 do IO<Result<&1, &1, U32 & String, Unit>>:
445 _closed : Unit <- File.close(file)
446 return Fail{(U32.from_nat(7n), "table exceeds packed write limit")}
447 case 1n+rest:
448 match state:
449 case WriteNext{file, encoder}:
450 write.step(rest, path,
451 WriteOutput{file, SstFile.next_chunk(encoder)})
452 case WriteOutput{file, pair}:
453 match pair:
454 case (_, None{}):
455 write.finish(file, path)
456 case (next, Some{bytes}):
457 do IO<Result<&1, &1, U32 & String, Unit>>:
458 written : File & Result<&1, &1, U32 & String, Unit> <-
459 Fs.write_bytes(file, bytes)
460 write.step(rest, path, WriteResult{written, next})
461 case WriteResult{pair, encoder}:
462 match pair:
463 case (file, Fail{error}):
464 do IO<Result<&1, &1, U32 & String, Unit>>:
465 _closed : Unit <- File.close(file)
466 return Fail{error}
467 case (file, Done{Unit{}}):
468 write.step(rest, path, WriteNext{file, encoder})
469
470def write.encoder(
471 +path: String,
472 fuel: Nat,
473 result: Result<&1, &1, SstFile.Error, SstFile.Encoder>
474) -> IO(Result<&1, &1, U32 & String, Unit>):
475 match result:
476 case Fail{error}:
477 IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{codec_error(error)})
478 case Done{encoder}:
479 do IO<Result<&1, &1, U32 & String, Unit>>:
480 file : File <- IO.try(File, File.open(path, "w"))
481 write.step(fuel, path, WriteNext{file, encoder})
482
483# Encode and write one SST as packed chunks, then sync it before publication.
484def write_table(
485 +path: String,
486 +entries: List<&2, MemTable.Entry>,
487 +level: U32
488) -> IO(Result<&1, &1, U32 & String, Unit>):
489 write.encoder(path,
490 Nat.add(Nat.mul(3n, Nat.add(List.length(&2, MemTable.Entry, entries), 2n)), 2n),
491 SstFile.new_encoder(entries, level))