src/DurableBatchPolicy.bend

mylsm-lsm-store@0.6.0.1 · 153 lines · 18 definitions

1import Base
2import mylsm-lsm-store@0.6.0.1/src/ByteWal.bend as ByteWal
3import mylsm-lsm-store@0.6.0.1/src/StorageBytes.bend as StorageBytes
4import mylsm-lsm-store@0.6.0.1/src/Wal.bend as Wal
5import mylsm-lsm-store@0.6.0.1/src/coordinator/QueuePolicy.bend as QueuePolicy
6import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
7
8# Carries the owned input and accumulated exact batch size through the fold.
9type ByteBatchMeasureState is Type:
10  MeasureNeed{mutations: List<&1, ByteWal.Mutation>,
11    reversed: List<&1, ByteWal.Mutation>, count: Nat, record_bytes: Nat}
12
13# Returns a measured list or rejects input exceeding the mutation-count bound.
14type ByteBatchMeasurement is Type:
15  BatchMeasured{mutations: List<&1, ByteWal.Mutation>, count: Nat,
16    encoded_bytes: Nat}
17  BatchTooMany{}
18
19# Returns the count-limit decision for a mutation batch.
20type Validation is Data:
21  ValidBatch{}
22  EmptyBatch{}
23  BatchCountExceeded{}
24  BatchBytesExceeded{}
25
26# Returns the maximum number of mutations in a batch.
27def max_batch_count() -> Nat:
28  256n
29
30# Caps an encoded multi-mutation WAL frame at the normal block size.
31def max_batch_bytes() -> Nat:
32  U32.to_nat(StorageBytes.MAX_BLOCK_BYTES())
33
34# Retains the codec's one-record exception to the normal frame limit.
35def max_single_batch_bytes() -> Nat:
36  Nat.add(U32.to_nat(StorageBytes.MAX_RECORD_BYTES()), 36n)
37
38# Selects the encoded-frame ceiling for a single record or mutation batch.
39def frame_byte_limit(
40  count: Nat,
41  batch_limit: Nat,
42  single_limit: Nat
43) -> Nat:
44  Bool.pick(Nat, Nat.is_eq(count, 1n), single_limit, batch_limit)
45
46# Counts one encoded Put record's tag, length fields, key, and value.
47def put_record_bytes(key_bytes: Nat, value_bytes: Nat) -> Nat:
48  Nat.add(9n, Nat.add(key_bytes, value_bytes))
49
50# Counts one encoded Delete record's tag, length field, and key.
51def delete_record_bytes(key_bytes: Nat) -> Nat:
52  Nat.add(5n, key_bytes)
53
54# Decides whether one encoded frame is within its selected byte ceiling.
55def validate_frame_bytes(
56  count: Nat,
57  encoded_bytes: Nat,
58  batch_limit: Nat,
59  single_limit: Nat
60) -> Validation:
61  Bool.pick(Validation,
62    Nat.is_le(encoded_bytes, frame_byte_limit(count, batch_limit, single_limit)),
63    ValidBatch{}, BatchBytesExceeded{})
64
65# Adds the fixed generation-four WAL frame header and checksum size.
66def frame_bytes(record_bytes: Nat) -> Nat:
67  Nat.add(36n, record_bytes)
68
69# Advances a bounded no-copy byte-batch size pass.
70def byte_batch.measure.loop(
71  fuel: Nat,
72  state: ByteBatchMeasureState
73) -> ByteBatchMeasurement:
74  match fuel:
75    case 0n:
76      match state:
77        case MeasureNeed{Nil{}, reversed, count, record_bytes}:
78          BatchMeasured{List.reverse(&1, ByteWal.Mutation, reversed), count,
79            frame_bytes(record_bytes)}
80        case MeasureNeed{Con{_, _}, _, _, _}: BatchTooMany{}
81    case 1n+rest:
82      match state:
83        case MeasureNeed{Nil{}, reversed, count, record_bytes}:
84          BatchMeasured{List.reverse(&1, ByteWal.Mutation, reversed), count,
85            frame_bytes(record_bytes)}
86        case MeasureNeed{Con{ByteWal.Delete{key}, tail}, reversed, count,
87            record_bytes}:
88          match key:
89            case Bytes.Bytes{+key_len, key_buffer}:
90              byte_batch.measure.loop(rest, MeasureNeed{tail,
91                Con{ByteWal.Delete{Bytes.Bytes{key_len, key_buffer}}, reversed},
92                Nat.add(count, 1n),
93                Nat.add(record_bytes,
94                  delete_record_bytes(U32.to_nat(key_len)))})
95        case MeasureNeed{Con{ByteWal.Put{key, value}, tail}, reversed, count,
96            record_bytes}:
97          match key:
98            case Bytes.Bytes{+key_len, key_buffer}:
99              match value:
100                case Bytes.Bytes{+value_len, value_buffer}:
101                  byte_batch.measure.loop(rest, MeasureNeed{tail,
102                    Con{ByteWal.Put{Bytes.Bytes{key_len, key_buffer},
103                      Bytes.Bytes{value_len, value_buffer}}, reversed},
104                    Nat.add(count, 1n), Nat.add(record_bytes,
105                      put_record_bytes(U32.to_nat(key_len),
106                        U32.to_nat(value_len)))})
107
108# Measures a byte batch without copying buffers or building its WAL frame.
109def byte_batch.measure(
110  mutations: List<&1, ByteWal.Mutation>
111) -> ByteBatchMeasurement:
112  byte_batch.measure.loop(max_batch_count(),
113    MeasureNeed{mutations, Nil{}, 0n, 0n})
114
115# Checks whether a batch count is within the allowed range.
116def within_batch_count(count: Nat) -> Bool:
117  Nat.is_le(count, max_batch_count())
118
119# Rejects empty and oversized batches before writing.
120def validate_count(count: Nat) -> Validation:
121  match count:
122    case 0n:
123      EmptyBatch{}
124    case count:
125      Bool.pick(Validation, within_batch_count(count), ValidBatch{}, BatchCountExceeded{})
126
127# Applies the byte limit only after the mutation count is accepted.
128def validate_batch.counted(
129  count: Nat,
130  encoded_bytes: Nat,
131  count_validation: Validation
132) -> Validation:
133  match count_validation:
134    case ValidBatch{}:
135      validate_frame_bytes(count, encoded_bytes, max_batch_bytes(),
136        max_single_batch_bytes())
137    case EmptyBatch{}: EmptyBatch{}
138    case BatchCountExceeded{}: BatchCountExceeded{}
139    case BatchBytesExceeded{}: BatchBytesExceeded{}
140
141# Rejects a batch before storage access if its count or frame bytes exceed limits.
142def validate_batch.copies(
143  counts: Nat & Nat,
144  encoded_bytes: Nat
145) -> Validation:
146  match counts:
147    case (for_frame_limit, for_count_validation):
148      validate_batch.counted(for_frame_limit, encoded_bytes,
149        validate_count(for_count_validation))
150
151# Rejects a batch before storage access if its count or frame bytes exceed limits.
152def validate_batch(count: Nat, encoded_bytes: Nat) -> Validation:
153  validate_batch.copies(QueuePolicy.nat.clone.pair(count), encoded_bytes)