src/DurableBatchPolicy.bend
mylsm-lsm-store@0.6.0.0 · 153 lines · 0 definitions
1import Base
2import mylsm-lsm-store@0.6.0.0/src/ByteWal.bend as ByteWal
3import mylsm-lsm-store@0.6.0.0/src/StorageBytes.bend as StorageBytes
4import mylsm-lsm-store@0.6.0.0/src/Wal.bend as Wal
5import mylsm-lsm-store@0.6.0.0/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)