src/coordinator/Operation.bend
mylsm-lsm-store@0.6.0.1 · 319 lines · 41 definitions
1import Base
2import mylsm-lsm-store@0.6.0.1/src/ByteMemTable.bend as ByteMemTable
3import mylsm-lsm-store@0.6.0.1/src/ByteWal.bend as ByteWal
4import mylsm-lsm-store@0.6.0.1/src/Wal.bend as Wal
5import mylsm-lsm-store@0.6.0.1/src/coordinator/Protocol.bend as Protocol
6import mylsm-lsm-store@0.6.0.1/src/coordinator/CursorOpenCodec.bend as CursorOpenCodec
7import mylsm-lsm-store@0.6.0.1/src/scan/RangeCursor.bend as RangeCursor
8import bend-kit-bytes@0.3.2.0/bytes.bend as Bytes
9
10
11type Operation is Type:
12 Get{key: Bytes.Bytes}
13 Put{key: Bytes.Bytes, value: Bytes.Bytes}
14 Delete{key: Bytes.Bytes}
15 WriteBatch{mutations: List<&1, ByteWal.Mutation>}
16 Flush{}
17 Compact{}
18 Stats{}
19 SnapshotOpen{}
20 SnapshotGet{snapshot_id: U32, key: Bytes.Bytes}
21 SnapshotClose{snapshot_id: U32}
22 SnapshotCursorOpen{snapshot_id: U32, bounds: RangeCursor.Bounds}
23 SnapshotCursorNext{cursor_id: U32}
24 SnapshotCursorClose{cursor_id: U32}
25 SnapshotCursorSeek{cursor_id: U32, lower: Maybe<&1, Bytes.Bytes>}
26 DescribeOwner{}
27
28
29type Decision is Type:
30 Decoded{operation: Operation}
31 MalformedArguments{}
32 UnsupportedOperation{}
33
34def bytes.triple.second(
35 first: Bytes.Bytes,
36 pair: Bytes.Bytes & Bytes.Bytes
37) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
38 match pair:
39 case (second, third): (first, second, third)
40
41def bytes.triple.pair(
42 pair: Bytes.Bytes & Bytes.Bytes
43) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
44 match pair:
45 case (first, second):
46 bytes.triple.second(first, ByteMemTable.bytes.clone(second))
47
48def bytes.triple(bytes: Bytes.Bytes) -> Bytes.Bytes & Bytes.Bytes & Bytes.Bytes:
49 bytes.triple.pair(ByteMemTable.bytes.clone(bytes))
50
51def put.result(
52 key: Bytes.Bytes & Bytes.Bytes,
53 value: Bytes.Bytes & Bytes.Bytes
54) -> Decision:
55 match key:
56 case (_, owned_key):
57 match value:
58 case (_, owned_value): Decoded{Put{owned_key, owned_value}}
59
60def put.sliced(
61 copies: Bytes.Bytes & Bytes.Bytes,
62 key_lengths: U32 & U32,
63 value_lengths: U32 & U32
64) -> Decision:
65 match copies:
66 case (key_source, value_source):
67 match key_lengths:
68 case (key_length, key_offset):
69 match value_lengths:
70 case (value_length, _):
71 put.result(Bytes.slice(key_source, 8, key_length),
72 Bytes.slice(value_source, (8 + key_offset : U32), value_length))
73
74def put.bound.result(
75 valid: Bool,
76 payload: Bytes.Bytes,
77 key_lengths: U32 & U32,
78 value_lengths: U32 & U32
79) -> Decision:
80 match valid:
81 case False{}: MalformedArguments{}
82 case True{}:
83 put.sliced(ByteMemTable.bytes.clone(payload), key_lengths, value_lengths)
84
85def put.bound.checked(
86 payload: Bytes.Bytes,
87 actual_length: U32,
88 key_lengths: U32 & U32 & U32,
89 value_lengths: U32 & U32 & U32
90) -> Decision:
91 match key_lengths:
92 case (key_check, (key_length, key_offset)):
93 match value_lengths:
94 case (value_check, (value_length, value_copy)):
95 put.bound.result(
96 Nat.is_eq(U32.to_nat(actual_length),
97 Nat.add(8n, Nat.add(U32.to_nat(key_check),
98 U32.to_nat(value_check)))),
99 payload, (key_length, key_offset), (value_length, value_copy))
100
101def put.measure(
102 pair: Bytes.Bytes & U32,
103 key_lengths: U32 & U32 & U32,
104 value_lengths: U32 & U32 & U32
105) -> Decision:
106 match pair:
107 case (payload, actual_length):
108 put.bound.checked(payload, actual_length, key_lengths, value_lengths)
109
110def put.value.parsed(
111 pair: Bytes.Bytes & Maybe<&2, U32>,
112 measure: Bytes.Bytes,
113 key_length: U32
114) -> Decision:
115 match pair:
116 case (_, Some{value_length}):
117 put.measure(Bytes.length(measure),
118 Protocol.u32.duplicate.triple(key_length),
119 Protocol.u32.duplicate.triple(value_length))
120 case _: MalformedArguments{}
121
122def put.value.length.keyed(
123 pair: Bytes.Bytes & Maybe<&2, U32>,
124 measure: Bytes.Bytes
125) -> Decision:
126 match pair:
127 case (value_head, Some{key_length}):
128 put.value.parsed(Bytes.get.u32be(value_head, 4), measure, key_length)
129 case _: MalformedArguments{}
130
131def put.decode.copies(copies: Bytes.Bytes & Bytes.Bytes & Bytes.Bytes) -> Decision:
132 match copies:
133 case (key_head, (value_head, measure)):
134 put.value.length.keyed(Bytes.get.u32be(key_head, 0), measure)
135
136def put.decode(payload: Bytes.Bytes) -> Decision:
137 put.decode.copies(bytes.triple(payload))
138
139def empty.result(
140 is_empty: Bool,
141 operation: Operation
142) -> Decision:
143 match is_empty:
144 case True{}: Decoded{operation}
145 case False{}: MalformedArguments{}
146
147def empty.operation(
148 pair: Bytes.Bytes & U32,
149 operation: Operation
150) -> Decision:
151 match pair:
152 case (_, length): empty.result(U32.is_eq(length, 0), operation)
153
154def batch.result(
155 decoded: Result<&1, &1, Wal.Error, ByteWal.Batch>
156) -> Decision:
157 match decoded:
158 case Fail{_}: MalformedArguments{}
159 case Done{ByteWal.Batch{mutations}}:
160 Decoded{WriteBatch{mutations}}
161
162def snapshot.get.key.sliced(pair: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes:
163 match pair:
164 case (_, key): key
165
166def snapshot.get.decision(snapshot_id: U32, key: Bytes.Bytes) -> Decision:
167 Decoded{SnapshotGet{snapshot_id, key}}
168
169def snapshot.get.decoded(
170 payload: Bytes.Bytes,
171 snapshot_id: U32,
172 length: U32
173) -> Decision:
174 snapshot.get.decision(snapshot_id,
175 snapshot.get.key.sliced(Bytes.slice(payload, 4, U32.sub(length, 4))))
176
177def snapshot.get.header.result(
178 payload: Bytes.Bytes,
179 snapshot_id: U32,
180 length: U32,
181 has_header: Bool
182) -> Decision:
183 match has_header:
184 case True{}: snapshot.get.decoded(payload, snapshot_id, length)
185 case False{}: MalformedArguments{}
186
187def snapshot.get.parsed(
188 parsed: Bytes.Bytes & Maybe<&2, U32>,
189 length: U32,
190 has_header: Bool
191) -> Decision:
192 match parsed:
193 case (payload, Some{snapshot_id}):
194 snapshot.get.header.result(payload, snapshot_id, length, has_header)
195 case _: MalformedArguments{}
196
197def snapshot.get.length.checked(
198 parsed: Bytes.Bytes & Maybe<&2, U32>,
199 lengths: U32 & U32 & U32
200) -> Decision:
201 match lengths:
202 case (length, (check_length, _)):
203 snapshot.get.parsed(parsed, length, U32.is_le(4, check_length))
204
205def snapshot.get.measured(measured: Bytes.Bytes & U32) -> Decision:
206 match measured:
207 case (payload, length):
208 snapshot.get.length.checked(Bytes.get.u32be(payload, 0),
209 Protocol.u32.duplicate.triple(length))
210
211def snapshot.get.decode(payload: Bytes.Bytes) -> Decision:
212 snapshot.get.measured(Bytes.length(payload))
213
214
215def snapshot.close.decision(parsed: Maybe<&2, U32>) -> Decision:
216 match parsed:
217 case Some{snapshot_id}: Decoded{SnapshotClose{snapshot_id}}
218 case None{}: MalformedArguments{}
219
220
221def snapshot.close.decode(payload: Bytes.Bytes) -> Decision:
222 snapshot.close.decision(Protocol.snapshot.id.read(payload))
223
224
225def snapshot.cursor.open.decision(
226 decision: CursorOpenCodec.Decision
227) -> Decision:
228 match decision:
229 case CursorOpenCodec.Decoded{CursorOpenCodec.Request{snapshot_id, bounds}}:
230 Decoded{SnapshotCursorOpen{snapshot_id, bounds}}
231 case CursorOpenCodec.Malformed{}: MalformedArguments{}
232
233
234def snapshot.cursor.open.decode(payload: Bytes.Bytes) -> Decision:
235 snapshot.cursor.open.decision(CursorOpenCodec.decode(payload))
236
237
238def snapshot.cursor.next.parsed(parsed: Maybe<&2, U32>) -> Decision:
239 match parsed:
240 case Some{cursor_id}: Decoded{SnapshotCursorNext{cursor_id}}
241 case None{}: MalformedArguments{}
242
243
244def snapshot.cursor.next.decode(payload: Bytes.Bytes) -> Decision:
245 snapshot.cursor.next.parsed(Protocol.snapshot.id.read(payload))
246
247
248def snapshot.cursor.close.parsed(parsed: Maybe<&2, U32>) -> Decision:
249 match parsed:
250 case Some{cursor_id}: Decoded{SnapshotCursorClose{cursor_id}}
251 case None{}: MalformedArguments{}
252
253
254def snapshot.cursor.close.decode(payload: Bytes.Bytes) -> Decision:
255 snapshot.cursor.close.parsed(Protocol.snapshot.id.read(payload))
256
257def snapshot.cursor.seek.range(
258 snapshot_id: U32,
259 bounds: RangeCursor.Bounds
260) -> Decision:
261 match bounds:
262 case RangeCursor.Bounds{lower, None{}, None{}}:
263 Decoded{SnapshotCursorSeek{snapshot_id, lower}}
264 case _: MalformedArguments{}
265
266def snapshot.cursor.seek.decision(
267 decision: CursorOpenCodec.Decision
268) -> Decision:
269 match decision:
270 case CursorOpenCodec.Decoded{CursorOpenCodec.Request{snapshot_id, bounds}}:
271 snapshot.cursor.seek.range(snapshot_id, bounds)
272 case CursorOpenCodec.Malformed{}: MalformedArguments{}
273
274def snapshot.cursor.seek.decode(payload: Bytes.Bytes) -> Decision:
275 snapshot.cursor.seek.decision(CursorOpenCodec.decode(payload))
276
277def decode.payload(
278 operation_code: U32,
279 payload: Bytes.Bytes
280) -> Decision:
281 match operation_code:
282 case 1: Decoded{Get{payload}}
283 case 2: put.decode(payload)
284 case 3: Decoded{Delete{payload}}
285 case 4: empty.operation(Bytes.length(payload), Flush{})
286 case 5: empty.operation(Bytes.length(payload), Compact{})
287 case 6: empty.operation(Bytes.length(payload), Stats{})
288 case 8: batch.result(ByteWal.decode(payload))
289 case 9: empty.operation(Bytes.length(payload), SnapshotOpen{})
290 case 10: snapshot.get.decode(payload)
291 case 11: snapshot.close.decode(payload)
292 case 12: snapshot.cursor.open.decode(payload)
293 case 13: snapshot.cursor.next.decode(payload)
294 case 14: snapshot.cursor.close.decode(payload)
295 case 15: snapshot.cursor.seek.decode(payload)
296 case 16: empty.operation(Bytes.length(payload), DescribeOwner{})
297 case _: UnsupportedOperation{}
298
299
300def decode(request: Protocol.Request) -> Decision:
301 match request:
302 case Protocol.Request{_, _, _, operation, payload}:
303 decode.payload(operation, payload)
304
305def u32.bytes(value: U32) -> Bytes.Bytes:
306 Bytes.set.u32be(Bytes.new(4), 0, value)
307
308def encode.put.sized(
309 key_pair: Bytes.Bytes & U32,
310 value_pair: Bytes.Bytes & U32
311) -> Bytes.Bytes:
312 match key_pair:
313 case (key, key_length):
314 match value_pair:
315 case (value, value_length):
316 Bytes.concat([u32.bytes(key_length), u32.bytes(value_length), key, value])
317
318def encode.put(key: Bytes.Bytes, value: Bytes.Bytes) -> Bytes.Bytes:
319 encode.put.sized(Bytes.length(key), Bytes.length(value))