src/CompactIo.bend source

src/CompactIo.bend on the hub · documented module

import Baseimport ./Compact.bend as Compactimport ./MemTable.bend as MemTableimport ./Sstable.bend as Sstableimport ./SstStreamIo.bend as SstStreamIoimport ./Wal.bend as Walimport ./Manifest.bend as Manifestimport ./Flush.bend as Flushimport ./Fs.bend as Fsimport ./Db.bend as Dbimport ./CrashPoint.bend as CrashPoint# Handle remove list in the compaction filesystem effects.def remove_list(names: List<&2, String>) -> IO(Result<&1, &1, U32 & String, Unit>):  match names:    case Nil{}:      IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})    case Con{n, t}:      do IO<Result<&1, &1, U32 & String, Unit>>:        _rm : Result<&1, &1, U32 & String, Unit> <- Fs.remove(n)        remove_list(t)# Tracks progress while removing published files.type RemoveState is Type:  RemoveStart{names: List<&2, String>}  RemoveNames{names: List<&2, String>}  RemoveResult{result: Result<&1, &1, U32 & String, Unit>, names: List<&2, String>}# Removes every path and propagates the first failure.def remove_list_strict(fuel: Nat, state: RemoveState) -> IO(Result<&1, &1, U32 & String, Unit>):  match fuel:    case 0n:      IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{(U32.from_nat(7n), "remove list exhausted")})    case 1n+rest:      match state:        case RemoveStart{names}:          remove_list_strict(rest, RemoveNames{names})        case RemoveNames{names}:          match names:            case Nil{}:              IO.pure(Result<&1, &1, U32 & String, Unit>, Done{Unit{}})            case Con{name, tail}:              do IO<Result<&1, &1, U32 & String, Unit>>:                removed : Result<&1, &1, U32 & String, Unit> <- Fs.remove(name)                remove_list_strict(rest, RemoveResult{removed, tail})        case RemoveResult{result, names}:          match result:            case Fail{error}:              IO.pure(Result<&1, &1, U32 & String, Unit>, Fail{error})            case Done{Unit{}}:              remove_list_strict(rest, RemoveNames{names})# Starts checked removal of the published file list.def remove_list_strict_start(pair: Nat & List<&2, String>) -> IO(Result<&1, &1, U32 & String, Unit>):  match pair:    case (fuel, names):      remove_list_strict(fuel, RemoveStart{names})def compact_goB.removed(  result: Result<&1, &1, U32 & String, Unit>,  +dir: String,  +mem: MemTable.MemTable,  +levels2: List<&2, List<&2, Sstable.Table>>,  +flushed: Nat,  +manifest: Manifest.Manifest,  +frozen: MemTable.MemTable,  +batch_cap: Nat,  +mem_count: Nat,  +frozen_count: Nat) -> IO(Result<&1, &1, U32 & String, Db.Db>):  match result:    case Fail{error}:      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})    case Done{Unit{}}:      IO.pure(Result<&1, &1, U32 & String, Db.Db>,        Done{Db.Db{dir, mem, frozen, batch_cap, Nil{}, levels2,          Nat.add(flushed, 1n), Manifest.token(manifest), mem_count, frozen_count}})def compact_goB.directory_synced(  result: Result<&1, &1, U32 & String, Unit>,  +dir: String,  +dels: List<&2, String>,  +m2: Manifest.Manifest,  +mem: MemTable.MemTable,  +levels2: List<&2, List<&2, Sstable.Table>>,  +flushed: Nat,  +frozen: MemTable.MemTable,  +batch_cap: Nat,  +mem_count: Nat,  +frozen_count: Nat) -> IO(Result<&1, &1, U32 & String, Db.Db>):  match result:    case Fail{error}:      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})    case Done{Unit{}}:      do IO<Result<&1, &1, U32 & String, Db.Db>>:        _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("compact.manifest_published"))        removed : Result<&1, &1, U32 & String, Unit> <- remove_list_strict_start((Nat.add(Nat.mul(2n, List.length(&2, String, dels)), 2n), dels))        compact_goB.removed(removed, dir, mem, levels2, flushed, m2,          frozen, batch_cap, mem_count, frozen_count)def compact_goB.renamed(  result: Result<&1, &1, U32 & String, Unit>,  +dir: String,  +dels: List<&2, String>,  +m2: Manifest.Manifest,  +mem: MemTable.MemTable,  +levels2: List<&2, List<&2, Sstable.Table>>,  +flushed: Nat,  +frozen: MemTable.MemTable,  +batch_cap: Nat,  +mem_count: Nat,  +frozen_count: Nat) -> IO(Result<&1, &1, U32 & String, Db.Db>):  match result:    case Fail{error}:      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})    case Done{Unit{}}:      do IO<Result<&1, &1, U32 & String, Db.Db>>:        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(dir)        compact_goB.directory_synced(synced, dir, dels, m2, mem, levels2,          flushed, frozen, batch_cap, mem_count, frozen_count)def compact_goB.written(  result: Result<&1, &1, U32 & String, Unit>,  +dir: String,  +mpath: String,  +mtmp: String,  +dels: List<&2, String>,  +m2: Manifest.Manifest,  +mem: MemTable.MemTable,  +levels2: List<&2, List<&2, Sstable.Table>>,  +flushed: Nat,  +frozen: MemTable.MemTable,  +batch_cap: Nat,  +mem_count: Nat,  +frozen_count: Nat) -> IO(Result<&1, &1, U32 & String, Db.Db>):  match result:    case Fail{error}:      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})    case Done{Unit{}}:      do IO<Result<&1, &1, U32 & String, Db.Db>>:        _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("compact.manifest_synced"))        renamed : Result<&1, &1, U32 & String, Unit> <- Fs.rename(mtmp, mpath)        compact_goB.renamed(renamed, dir, dels, m2, mem,          levels2, flushed, frozen, batch_cap, mem_count, frozen_count)# Compact goB for the compaction filesystem effects.def compact_goB(  +dir: String,  +mpath: String,  +mtmp: String,  +m2: Manifest.Manifest,  +dels: List<&2, String>,  +mem: MemTable.MemTable,  +levels2: List<&2, List<&2, Sstable.Table>>,  +flushed: Nat,  +frozen: MemTable.MemTable,  +batch_cap: Nat,  +mem_count: Nat,  +frozen_count: Nat,) -> IO(Result<&1, &1, U32 & String, Db.Db>):  do IO<Result<&1, &1, U32 & String, Db.Db>>:    written : Result<&1, &1, U32 & String, Unit> <- Flush.write_manifest(mtmp, m2)    compact_goB.written(written, dir, mpath, mtmp, dels, m2, mem, levels2,      flushed, frozen, batch_cap, mem_count, frozen_count)# Handle drift go in the compaction filesystem effects.def drift_go(  ok: Bool,  +dir: String,  +mpath: String,  +mtmp: String,  +mfst: Manifest.Manifest,  +oldl1: List<&2, Sstable.Table>,  +remt: List<&2, Sstable.Table>,  +outname: String,  +mem: MemTable.MemTable,  +levels2: List<&2, List<&2, Sstable.Table>>,  +flushed: Nat,  +frozen: MemTable.MemTable,  +batch_cap: Nat,  +mem_count: Nat,  +frozen_count: Nat,) -> IO(Result<&1, &1, U32 & String, Db.Db>):  match ok:    case True{}:      +new_mfst = Compact.mfst_new(mfst, Compact.rem_names(oldl1, Compact.ml1(mfst), remt), outname)      +removed = Compact.abs_names(oldl1, Compact.ml1(mfst), remt)      +dels = List.append(&2, String, Compact.prefix(Compact.ml0(mfst), dir ++ "/l0"), Compact.prefix(removed, dir ++ "/l1"))      compact_goB(dir, mpath, mtmp, new_mfst, dels, mem, levels2, flushed, frozen, batch_cap, mem_count, frozen_count)    case False{}:      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{(U32.from_nat(2n), "manifest drift")})def compact_goA.loaded(  result: Result<&1, &1, U32 & String, Manifest.Manifest>,  +dir: String,  +mpath: String,  +mtmp: String,  +oldl1: List<&2, Sstable.Table>,  +remt: List<&2, Sstable.Table>,  +outname: String,  +mem: MemTable.MemTable,  +levels2: List<&2, List<&2, Sstable.Table>>,  +flushed: Nat,  +manifest_token: String,  +frozen: MemTable.MemTable,  +batch_cap: Nat,  +mem_count: Nat,  +frozen_count: Nat) -> IO(Result<&1, &1, U32 & String, Db.Db>):  match result:    case Fail{error}:      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})    case Done{+manifest}:      drift_go(String.eq(Manifest.token(manifest), manifest_token), dir, mpath,        mtmp, manifest, oldl1, remt, outname, mem, levels2, flushed,        frozen, batch_cap, mem_count, frozen_count)def compact_goA.synced(  result: Result<&1, &1, U32 & String, Unit>,  +dir: String,  +mpath: String,  +mtmp: String,  +oldl1: List<&2, Sstable.Table>,  +remt: List<&2, Sstable.Table>,  +outname: String,  +mem: MemTable.MemTable,  +levels2: List<&2, List<&2, Sstable.Table>>,  +flushed: Nat,  +manifest_token: String,  +frozen: MemTable.MemTable,  +batch_cap: Nat,  +mem_count: Nat,  +frozen_count: Nat) -> IO(Result<&1, &1, U32 & String, Db.Db>):  match result:    case Fail{error}:      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})    case Done{Unit{}}:      do IO<Result<&1, &1, U32 & String, Db.Db>>:        _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("compact.output_published"))        manifest : Result<&1, &1, U32 & String, Manifest.Manifest> <- Flush.load_mfst(mpath)        compact_goA.loaded(manifest, dir, mpath, mtmp, oldl1, remt, outname,          mem, levels2, flushed, manifest_token, frozen, batch_cap, mem_count, frozen_count)def compact_goA.renamed(  result: Result<&1, &1, U32 & String, Unit>,  +dir: String,  +l1dir: String,  +mpath: String,  +mtmp: String,  +oldl1: List<&2, Sstable.Table>,  +remt: List<&2, Sstable.Table>,  +outname: String,  +mem: MemTable.MemTable,  +levels2: List<&2, List<&2, Sstable.Table>>,  +flushed: Nat,  +manifest_token: String,  +frozen: MemTable.MemTable,  +batch_cap: Nat,  +mem_count: Nat,  +frozen_count: Nat) -> IO(Result<&1, &1, U32 & String, Db.Db>):  match result:    case Fail{error}:      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})    case Done{Unit{}}:      do IO<Result<&1, &1, U32 & String, Db.Db>>:        synced : Result<&1, &1, U32 & String, Unit> <- Fs.fsync(l1dir)        compact_goA.synced(synced, dir, mpath, mtmp, oldl1, remt, outname,          mem, levels2, flushed, manifest_token, frozen, batch_cap, mem_count, frozen_count)def compact_goA.written(  result: Result<&1, &1, U32 & String, Unit>,  +dir: String,  +l1dir: String,  +tmp: String,  +final: String,  +mpath: String,  +mtmp: String,  +oldl1: List<&2, Sstable.Table>,  +remt: List<&2, Sstable.Table>,  +outname: String,  +mem: MemTable.MemTable,  +levels2: List<&2, List<&2, Sstable.Table>>,  +flushed: Nat,  +manifest_token: String,  +frozen: MemTable.MemTable,  +batch_cap: Nat,  +mem_count: Nat,  +frozen_count: Nat) -> IO(Result<&1, &1, U32 & String, Db.Db>):  match result:    case Fail{error}:      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})    case Done{Unit{}}:      do IO<Result<&1, &1, U32 & String, Db.Db>>:        _checkpoint : Unit <- IO.try(Unit, CrashPoint.hit("compact.output_synced"))        renamed : Result<&1, &1, U32 & String, Unit> <- Fs.rename(tmp, final)        compact_goA.renamed(renamed, dir, l1dir, mpath, mtmp, oldl1, remt,          outname, mem, levels2, flushed, manifest_token, frozen, batch_cap, mem_count, frozen_count)def compact_goA.ensured(  result: Result<&1, &1, U32 & String, Unit>,  +dir: String,  +l1dir: String,  +tmp: String,  +final: String,  +mpath: String,  +mtmp: String,  +entries: List<&2, MemTable.Entry>,  +level: U32,  +outname: String,  +mem: MemTable.MemTable,  +levels2: List<&2, List<&2, Sstable.Table>>,  +oldl1: List<&2, Sstable.Table>,  +remt: List<&2, Sstable.Table>,  +flushed: Nat,  +manifest_token: String,  +frozen: MemTable.MemTable,  +batch_cap: Nat,  +mem_count: Nat,  +frozen_count: Nat) -> IO(Result<&1, &1, U32 & String, Db.Db>):  match result:    case Fail{error}:      IO.pure(Result<&1, &1, U32 & String, Db.Db>, Fail{error})    case Done{Unit{}}:      do IO<Result<&1, &1, U32 & String, Db.Db>>:        written : Result<&1, &1, U32 & String, Unit> <- SstStreamIo.write_table(tmp, entries, level)        compact_goA.written(written, dir, l1dir, tmp, final, mpath, mtmp,          oldl1, remt, outname, mem, levels2, flushed, manifest_token, frozen,          batch_cap, mem_count, frozen_count)# Publishes the compacted run and removes obsolete files.def compact_goA(  +dir: String,  +l1dir: String,  +tmp: String,  +final: String,  +mpath: String,  +mtmp: String,  +entries: List<&2, MemTable.Entry>,  +level: U32,  +outname: String,  +mem: MemTable.MemTable,  +levels2: List<&2, List<&2, Sstable.Table>>,  +oldl1: List<&2, Sstable.Table>,  +remt: List<&2, Sstable.Table>,  +flushed: Nat,  +manifest_token: String,  +frozen: MemTable.MemTable,  +batch_cap: Nat,  +mem_count: Nat,  +frozen_count: Nat,) -> IO(Result<&1, &1, U32 & String, Db.Db>):  do IO<Result<&1, &1, U32 & String, Db.Db>>:    ensured : Result<&1, &1, U32 & String, Unit> <- Flush.ensure_dir(l1dir)    compact_goA.ensured(ensured, dir, l1dir, tmp, final, mpath, mtmp,      entries, level, outname, mem, levels2, oldl1, remt, flushed,      manifest_token, frozen, batch_cap, mem_count, frozen_count)# Compact pre for the compaction filesystem effects.def compact_pre(  +dir: String,  +mem: MemTable.MemTable,  +levels: List<&2, List<&2, Sstable.Table>>,  +flushed: Nat,  +manifest_token: String,  +frozen: MemTable.MemTable,  +batch_cap: Nat,  +mem_count: Nat,  +frozen_count: Nat,) -> IO(Result<&1, &1, U32 & String, Db.Db>):  +levels2 = Compact.compact_levels(levels)  +newl1 = Compact.l1list(levels2)  +out = Compact.hd_tbl(newl1)  +remt = Compact.tail_of(newl1)  +l1dir = dir ++ "/l1"  +outname = "l1-" ++ Nat.show(flushed) ++ ".tbl"  +tmp = l1dir ++ "/" ++ outname ++ ".tmp"  +final = l1dir ++ "/" ++ outname  compact_goA(    dir, l1dir, tmp, final, dir ++ "/MANIFEST", dir ++ "/MANIFEST.tmp",    Db.table_entries(out), 1, outname, mem, levels2, Compact.l1list(levels), remt, flushed,    manifest_token, frozen, batch_cap, mem_count, frozen_count,  )# Compact trig for the compaction filesystem effects.def compact_trig(  gt: Bool,  +dir: String,  +mem: MemTable.MemTable,  +levels: List<&2, List<&2, Sstable.Table>>,  +flushed: Nat,  +manifest_token: String,  +frozen: MemTable.MemTable,  +batch_cap: Nat,  +bcache: List<&2, Db.BEntry>,  +mem_count: Nat,  +frozen_count: Nat,) -> IO(Result<&1, &1, U32 & String, Db.Db>):  match gt:    case True{}:      compact_pre(dir, mem, levels, flushed, manifest_token, frozen, batch_cap, mem_count, frozen_count)    case False{}:      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}})# Handle compact in the compaction filesystem effects.def compact(+db: Db.Db) -> IO(Result<&1, &1, U32 & String, Db.Db>):  match db:    case Db.Db{dir, mem, frozen, batch_cap, bcache, levels, flushed, manifest_token, mem_count, frozen_count}:      match levels:        case Nil{}:          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}})        case Con{+l0, r1}:          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)