redis.bend
bend-kit-redis@0.3.0.0 · 1,074 lines · 0 definitions
1# Redis and Valkey client: RESP3 over TCP or TLS, with pipelining and a pool. Source: https://github.com/paymog/bend-kit/tree/main/redis
2import Base
3import bend-kit-bytes@0.3.3.0/bytes.bend as Bytes
4import bend-kit-wire@0.4.6.2/wire.bend as Wire
5import bend-kit-dns@0.7.0.0/dns.bend as Dns
6
7# A RESP3 value (RESP2 replies decode into the same type). Text fields hold the raw
8# bytes of the line. Dict (a RESP3 map) and Attr keep their pairs flat, key then value, as
9# on the wire and in hiredis. Attr holds the attribute pairs, then the reply they annotate, last.
10# Int is a signed 64-bit integer as two's-complement words.
11type Val is Type:
12 Simple{text: Bytes.Bytes}
13 Error{text: Bytes.Bytes}
14 Int{hi: U32, lo: U32}
15 Bulk{data: Bytes.Bytes}
16 Arr{items: List<&1, Val>}
17 Null{}
18 Boolean{value: Bool}
19 Double{text: Bytes.Bytes}
20 BigNum{text: Bytes.Bytes}
21 BulkError{text: Bytes.Bytes}
22 Verbatim{format: Bytes.Bytes, text: Bytes.Bytes}
23 Dict{items: List<&1, Val>}
24 Set{items: List<&1, Val>}
25 Attr{items: List<&1, Val>}
26 Push{items: List<&1, Val>}
27
28# Aggregate kinds, in the order of their type bytes below.
29def kind.tag(+kind: U32) -> String:
30 match kind:
31 case 0:
32 "*"
33 case 1:
34 "%"
35 case 2:
36 "~"
37 case 3:
38 "|"
39 case _:
40 ">"
41
42def kind.val(+kind: U32, xs: List<&1, Val>) -> Val:
43 match kind:
44 case 0:
45 Arr{xs}
46 case 1:
47 Dict{xs}
48 case 2:
49 Set{xs}
50 case 3:
51 Attr{xs}
52 case _:
53 Push{xs}
54
55# ---- integers
56
57def i64.neg(+hi: U32, +lo: U32) -> U32 & U32:
58 (((hi .^. 4294967295) + Bool.pick(U32, U32.is_eq(lo, 0), 1, 0) : U32), ((lo .^. 4294967295) + 1 : U32))
59
60# The 64-bit (hi, lo) divided by 10, and the remainder, over 16-bit limbs.
61def u64.div10(+hi: U32, +lo: U32) -> U32 & U32 & U32:
62 +a = (hi >> 16n : U32)
63 +x = (((a % 10) << 16n) .|. (hi .&. 65535) : U32)
64 +y = (((x % 10) << 16n) .|. (lo >> 16n) : U32)
65 +z = (((y % 10) << 16n) .|. (lo .&. 65535) : U32)
66 ((((a / 10) << 16n) .|. (x / 10) : U32), (((y / 10) << 16n) .|. (z / 10) : U32), (z % 10 : U32))
67
68def digit.put(z: Bool, +r: U32, acc: String) -> String:
69 match z:
70 case True{}:
71 acc
72 case False{}:
73 SCon{Chr{(48 + r : U32)}, acc}
74
75# z is set once the value left to print is zero; 20 digits hold any u64.
76def u64.show.go(f: Nat, q: U32 & U32 & U32, acc: String, +z: Bool) -> String:
77 match f:
78 case 0n:
79 acc
80 case 1n+p:
81 # A + on a later tuple binder fails to check (https://github.com/bendlang/bend/issues/1077).
82 (hi0, lo0, r) = q
83 +hi = hi0
84 +lo = lo0
85 u64.show.go(p, u64.div10(hi, lo), digit.put(z, r, acc), Bool.or(z, U32.is_eq((hi .|. lo : U32), 0)))
86
87def u64.show(+hi: U32, +lo: U32) -> String:
88 u64.show.go(20n, u64.div10(hi, lo), SNil{}, False{})
89
90def i64.show.of(r: U32 & U32, neg: Bool) -> String:
91 (+hi, +lo) = r
92 match neg:
93 case True{}:
94 "-" ++ u64.show(hi, lo)
95 case False{}:
96 u64.show(hi, lo)
97
98def i64.show(+hi: U32, +lo: U32) -> String:
99 +neg = U32.is_le(2147483648, hi)
100 i64.show.of(Bool.pick(U32 & U32, neg, i64.neg(hi, lo), (hi, lo)), neg)
101
102# ---- encoder
103
104# The output so far, whether every value had a RESP form, and the count of values written.
105type E is Type:
106 E{o: Bytes.Bytes, ok: Bool, n: U32}
107
108def fresh() -> E:
109 E{Bytes.new(0), True{}, 0}
110
111def text(+s: String) -> Bytes.Bytes:
112 Bytes.from_string(s)
113
114def none(m: Maybe<&2, U32>) -> Bool:
115 match m:
116 case None{}:
117 True{}
118 case Some{x}:
119 False{}
120
121def clean.lf(+cr: Bool, r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & Bool:
122 (b, m) = r
123 (b, Bool.and(cr, none(m)))
124
125def clean.cr(r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & Bool:
126 (b, m) = r
127 clean.lf(none(m), Bytes.find(b, "\n"))
128
129# A line has no CR and no LF.
130def clean(t: Bytes.Bytes) -> Bytes.Bytes & Bool:
131 clean.cr(Bytes.find(t, "\r"))
132
133def put(st: E, +head: String, body: Bytes.Bytes, +good: Bool) -> E:
134 E{o, +ok, +n} = st
135 E{Bytes.append(Bytes.append(Bytes.append(o, text(head)), body), text("\r\n")), Bool.and(ok, good), (n + 1 : U32)}
136
137def line(st: E, +tag: String, r: Bytes.Bytes & Bool) -> E:
138 (t, +good) = r
139 put(st, tag, t, good)
140
141def blob(st: E, +tag: String, b: Bytes.Bytes) -> E:
142 Bytes.Bytes{+len, buf} = b
143 put(st, tag ++ U32.show(len) ++ "\r\n", Bytes.Bytes{len, buf}, True{})
144
145def verbatim.of(st: E, +tl: U32, r: Bytes.Bytes & U32, t: Bytes.Bytes) -> E:
146 (f, +fl) = r
147 put(st, "=" ++ U32.show((tl + 4 : U32)) ++ "\r\n", Bytes.append(Bytes.append(f, text(":")), t), U32.is_eq(fl, 3))
148
149def verbatim(st: E, f: Bytes.Bytes, t: Bytes.Bytes) -> E:
150 Bytes.Bytes{+tl, tb} = t
151 verbatim.of(st, tl, Bytes.length(f), Bytes.Bytes{tl, tb})
152
153# Maps count pairs and need an even count; Attr needs its pairs and one reply.
154def agg.count(+kind: U32, +n: U32) -> U32:
155 Bool.pick(U32, U32.is_eq(kind, 1), (n >> 1n : U32), Bool.pick(U32, U32.is_eq(kind, 3), (n >> 1n : U32), n))
156
157def agg.ok(+kind: U32, +n: U32) -> Bool:
158 +odd = U32.is_eq((n .&. 1 : U32), 1)
159 Bool.pick(Bool, U32.is_eq(kind, 1), Bool.not(odd), Bool.pick(Bool, U32.is_eq(kind, 3), odd, True{}))
160
161def agg(st: E, +kind: U32, sub: E) -> E:
162 E{o, +ok, +n} = st
163 E{body, +good, +k} = sub
164 +head = kind.tag(kind) ++ U32.show(agg.count(kind, k)) ++ "\r\n"
165 E{Bytes.append(Bytes.append(o, text(head)), body), Bool.and(ok, Bool.and(good, agg.ok(kind, k))), (n + 1 : U32)}
166
167def boolean(v: Bool) -> String:
168 match v:
169 case True{}:
170 "t"
171 case False{}:
172 "f"
173
174def enc(xs: List<&1, Val>, st: E) -> E:
175 match xs:
176 case Nil{}:
177 st
178 case Con{Simple{t}, rest}:
179 enc(rest, line(st, "+", clean(t)))
180 case Con{Error{t}, rest}:
181 enc(rest, line(st, "-", clean(t)))
182 case Con{Int{+hi, +lo}, rest}:
183 enc(rest, put(st, ":" ++ i64.show(hi, lo), Bytes.new(0), True{}))
184 case Con{Bulk{d}, rest}:
185 enc(rest, blob(st, "$", d))
186 case Con{Arr{items}, rest}:
187 enc(rest, agg(st, 0, enc(items, fresh())))
188 case Con{Null{}, rest}:
189 enc(rest, put(st, "_", Bytes.new(0), True{}))
190 case Con{Boolean{v}, rest}:
191 enc(rest, put(st, "#" ++ boolean(v), Bytes.new(0), True{}))
192 case Con{Double{t}, rest}:
193 enc(rest, line(st, ",", clean(t)))
194 case Con{BigNum{t}, rest}:
195 enc(rest, line(st, "(", clean(t)))
196 case Con{BulkError{t}, rest}:
197 enc(rest, blob(st, "!", t))
198 case Con{Verbatim{f, t}, rest}:
199 enc(rest, verbatim(st, f, t))
200 case Con{Dict{items}, rest}:
201 enc(rest, agg(st, 1, enc(items, fresh())))
202 case Con{Set{items}, rest}:
203 enc(rest, agg(st, 2, enc(items, fresh())))
204 case Con{Attr{items}, rest}:
205 enc(rest, agg(st, 3, enc(items, fresh())))
206 case Con{Push{items}, rest}:
207 enc(rest, agg(st, 4, enc(items, fresh())))
208
209def encode.done(st: E) -> Maybe<&1, Bytes.Bytes>:
210 E{o, ok, n} = st
211 match ok:
212 case True{}:
213 Some{o}
214 case False{}:
215 None{}
216
217# None when a line value holds CR or LF, a Dict has an odd item count, an Attr an even
218# one, or a Verbatim format is not 3 bytes.
219def encode(v: Val) -> Maybe<&1, Bytes.Bytes>:
220 encode.done(enc([v], fresh()))
221
222def bulks(xs: List<&1, Bytes.Bytes>) -> List<&1, Val>:
223 match xs:
224 case Nil{}:
225 Nil{}
226 case Con{h, t}:
227 Con{Bulk{h}, bulks(t)}
228
229def encode.command.done(st: E) -> Bytes.Bytes:
230 E{o, ok, n} = st
231 o
232
233# A command as a RESP array of bulk strings, the form every server accepts.
234def encode.command(args: List<&1, Bytes.Bytes>) -> Bytes.Bytes:
235 encode.command.done(enc([Arr{bulks(args)}], fresh()))
236
237# Byte-string arguments (one Char per octet) as Bytes.
238def args(xs: List<&2, String>) -> List<&1, Bytes.Bytes>:
239 match xs:
240 case Nil{}:
241 Nil{}
242 case Con{h, t}:
243 Con{Bytes.from_string(h), args(t)}
244
245# ---- decoder
246
247# A decimal line: sign, magnitude words, and whether it has 1 to 19 digits.
248type Num is Data:
249 Num{neg: Bool, hi: U32, lo: U32, ok: Bool}
250
251# n digits from i: (hi, lo) = (hi, lo) * 10 + d over 16-bit limbs. 19 digits stay below 2^64.
252def num.go(n: Nat, r: Array<U32> & U32, +i: U32, +hi: U32, +lo: U32, +ok: Bool) -> Array<U32> & U32 & U32 & Bool:
253 match n:
254 case 0n:
255 (a, c) = r
256 (a, hi, lo, ok)
257 case 1n+p:
258 (a, +c) = r
259 +d = (c - 48 : U32)
260 +p0 = ((lo .&. 65535) * 10 + d : U32)
261 +p1 = ((lo >> 16n) * 10 + (p0 >> 16n) : U32)
262 num.go(p, Bytes.peek(a, (i + 1 : U32)), (i + 1 : U32), (hi * 10 + (p1 >> 16n) : U32), ((p1 << 16n) .|. (p0 .&. 65535) : U32), Bool.and(ok, U32.is_lt(d, 10)))
263
264def num.fin(+neg: Bool, +len: U32, r: Array<U32> & U32 & U32 & Bool) -> Bytes.Bytes & Num:
265 (a, hi, lo, ok) = r
266 (Bytes.Bytes{len, a}, Num{neg, hi, lo, ok})
267
268def num.digits(+neg: Bool, a: Array<U32>, +len: U32, +s: U32, +e: U32) -> Bytes.Bytes & Num:
269 +k = (e - s : U32)
270 num.fin(neg, len, num.go(U32.to_nat(k), Bytes.peek(a, s), s, 0, 0, Bool.and(U32.is_lt(0, k), U32.is_le(k, 19))))
271
272def num.sign(+len: U32, +s: U32, +e: U32, r: Array<U32> & U32) -> Bytes.Bytes & Num:
273 (a, +c) = r
274 +neg = Bool.and(U32.is_eq(c, 45), U32.is_lt(s, e))
275 num.digits(neg, a, len, Bool.pick(U32, neg, (s + 1 : U32), s), e)
276
277# Bytes s until e, which are before len.
278def num(b: Bytes.Bytes, +s: U32, +e: U32) -> Bytes.Bytes & Num:
279 Bytes.Bytes{+len, buf} = b
280 num.sign(len, s, e, Bytes.peek(buf, s))
281
282def int.mk(r: U32 & U32) -> Val:
283 (hi, lo) = r
284 Int{hi, lo}
285
286def int.val(good: Bool, +neg: Bool, +hi: U32, +lo: U32) -> Maybe<&1, Val>:
287 match good:
288 case True{}:
289 Some{int.mk(Bool.pick(U32 & U32, neg, i64.neg(hi, lo), (hi, lo)))}
290 case False{}:
291 None{}
292
293# RESP integers are signed 64-bit: the magnitude may reach 2^63 only when negative.
294def int.of(n: Num) -> Maybe<&1, Val>:
295 Num{+neg, +hi, +lo, ok} = n
296 +fits = Bool.or(U32.is_lt(hi, 2147483648), Bool.and(neg, Bool.and(U32.is_eq(hi, 2147483648), U32.is_eq(lo, 0))))
297 int.val(Bool.and(ok, fits), neg, hi, lo)
298
299# A length or count below 2^31; -1 (RESP2 null) is 4294967295 and anything else is 4294967294.
300def len.of(n: Num) -> U32:
301 Num{+neg, +hi, +lo, +ok} = n
302 +nil = Bool.and(ok, Bool.and(neg, Bool.and(U32.is_eq(hi, 0), U32.is_eq(lo, 1))))
303 +good = Bool.and(ok, Bool.and(Bool.not(neg), Bool.and(U32.is_eq(hi, 0), U32.is_lt(lo, 2147483648))))
304 Bool.pick(U32, good, lo, Bool.pick(U32, nil, 4294967295, 4294967294))
305
306# An open aggregate: its kind, the items still due, and the items so far, reversed.
307type Frame is Type:
308 Frame{kind: U32, left: U32, xs: List<&1, Val>}
309
310type St is Type:
311 SHead{i: U32, stack: List<&1, Frame>}
312 SDone{v: Val, i: U32, stack: List<&1, Frame>}
313 SOk{v: Val, i: U32}
314 SMore{}
315 SBad{}
316
317# One reply and the index after it; More when the bytes end first; Bad when they cannot start a reply.
318type Step is Type:
319 Reply{v: Val, next: U32}
320 More{}
321 Bad{}
322
323def byte.of(r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & U32:
324 (b, m) = r
325 match m:
326 case Some{c}:
327 (b, c)
328 case None{}:
329 (b, 256)
330
331# Byte i, or 256 past the end.
332def byte(b: Bytes.Bytes, +i: U32) -> Bytes.Bytes & U32:
333 byte.of(Bytes.get(b, i))
334
335def known(+t: U32) -> Bool:
336 match t:
337 case 43:
338 True{}
339 case 45:
340 True{}
341 case 58:
342 True{}
343 case 36:
344 True{}
345 case 42:
346 True{}
347 case 95:
348 True{}
349 case 35:
350 True{}
351 case 44:
352 True{}
353 case 40:
354 True{}
355 case 33:
356 True{}
357 case 61:
358 True{}
359 case 37:
360 True{}
361 case 126:
362 True{}
363 case 124:
364 True{}
365 case 62:
366 True{}
367 case _:
368 False{}
369
370def null.st(ok: Bool, +next: U32, stack: List<&1, Frame>) -> St:
371 match ok:
372 case True{}:
373 SDone{Null{}, next, stack}
374 case False{}:
375 SBad{}
376
377def line.val(+k: U32, t: Bytes.Bytes) -> Val:
378 match k:
379 case 0:
380 Simple{t}
381 case 1:
382 Error{t}
383 case 2:
384 Double{t}
385 case _:
386 BigNum{t}
387
388def line.st(+k: U32, +next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes & St:
389 (b, t) = r
390 (b, SDone{line.val(k, t), next, stack})
391
392def int.done(m: Maybe<&1, Val>, +next: U32, stack: List<&1, Frame>) -> St:
393 match m:
394 case Some{v}:
395 SDone{v, next, stack}
396 case None{}:
397 SBad{}
398
399def int.st(+next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Num) -> Bytes.Bytes & St:
400 (b, n) = r
401 (b, int.done(int.of(n), next, stack))
402
403def boolean.done(ok: Bool, +v: Bool, +next: U32, stack: List<&1, Frame>) -> St:
404 match ok:
405 case True{}:
406 SDone{Boolean{v}, next, stack}
407 case False{}:
408 SBad{}
409
410def boolean.st(+next: U32, stack: List<&1, Frame>, +one: Bool, r: Bytes.Bytes & U32) -> Bytes.Bytes & St:
411 (b, +c) = r
412 (b, boolean.done(Bool.and(one, Bool.or(U32.is_eq(c, 116), U32.is_eq(c, 102))), U32.is_eq(c, 116), next, stack))
413
414def blob.mk(+kind: U32, d: Bytes.Bytes) -> Val:
415 match kind:
416 case 0:
417 Bulk{d}
418 case _:
419 BulkError{d}
420
421def blob.done(+kind: U32, +after: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes & St:
422 (b, d) = r
423 (b, SDone{blob.mk(kind, d), after, stack})
424
425def verb.text(f: Bytes.Bytes, +after: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes & St:
426 (b, t) = r
427 (b, SDone{Verbatim{f, t}, after, stack})
428
429def verb.fmt(+n: U32, +next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Bytes.Bytes) -> Bytes.Bytes & St:
430 (b, f) = r
431 verb.text(f, (next + n + 2 : U32), stack, Bytes.slice(b, (next + 4 : U32), (n - 4 : U32)))
432
433def verb.ok(ok: Bool, +n: U32, +next: U32, stack: List<&1, Frame>, b: Bytes.Bytes) -> Bytes.Bytes & St:
434 match ok:
435 case True{}:
436 verb.fmt(n, next, stack, Bytes.slice(b, next, 3))
437 case False{}:
438 (b, SBad{})
439
440def verb.colon(+n: U32, +next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & U32) -> Bytes.Bytes & St:
441 (b, +c) = r
442 verb.ok(U32.is_eq(c, 58), n, next, stack, b)
443
444# A verbatim string is a 3-byte format, a colon, then the text.
445def verb.len(ok: Bool, +n: U32, +next: U32, stack: List<&1, Frame>, b: Bytes.Bytes) -> Bytes.Bytes & St:
446 match ok:
447 case True{}:
448 verb.colon(n, next, stack, byte(b, (next + 3 : U32)))
449 case False{}:
450 (b, SBad{})
451
452def blob.val(+kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>, b: Bytes.Bytes) -> Bytes.Bytes & St:
453 match kind:
454 case 2:
455 verb.len(U32.is_le(4, n), n, next, stack, b)
456 case _:
457 blob.done(kind, (next + n + 2 : U32), stack, Bytes.slice(b, next, n))
458
459def blob.crlf(+kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Bool) -> Bytes.Bytes & St:
460 (b, ok) = r
461 match ok:
462 case True{}:
463 blob.val(kind, n, next, stack, b)
464 case False{}:
465 (b, SBad{})
466
467def crlf.at(b: Bytes.Bytes, +j: U32) -> Bytes.Bytes & Bool:
468 Bytes.Bytes{+len, buf} = b
469 Bytes.at.fin(len, Bytes.at(buf, "\r\n", j))
470
471def blob.fit(ok: Bool, +kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>, b: Bytes.Bytes) -> Bytes.Bytes & St:
472 match ok:
473 case True{}:
474 blob.crlf(kind, n, next, stack, crlf.at(b, (next + n : U32)))
475 case False{}:
476 (b, SMore{})
477
478def blob.room(+kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & U32) -> Bytes.Bytes & St:
479 (b, +len) = r
480 blob.fit(Bytes.fits(len, next, (n + 2 : U32)), kind, n, next, stack, b)
481
482def blob.bad(bad: Bool, +kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>, b: Bytes.Bytes) -> Bytes.Bytes & St:
483 match bad:
484 case True{}:
485 (b, SBad{})
486 case False{}:
487 blob.room(kind, n, next, stack, Bytes.length(b))
488
489# $-1 is RESP2's null bulk string; ! and = have no null form.
490def blob.nil(nil: Bool, +kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>, b: Bytes.Bytes) -> Bytes.Bytes & St:
491 match nil:
492 case True{}:
493 (b, null.st(U32.is_eq(kind, 0), next, stack))
494 case False{}:
495 blob.bad(U32.is_eq(n, 4294967294), kind, n, next, stack, b)
496
497# kind 0 is $, 1 is !, 2 is =.
498def blob.st(+kind: U32, +next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Num) -> Bytes.Bytes & St:
499 (b, n) = r
500 +k = len.of(n)
501 blob.nil(U32.is_eq(k, 4294967295), kind, k, next, stack, b)
502
503def agg.open(empty: Bool, +kind: U32, +left: U32, +next: U32, stack: List<&1, Frame>) -> St:
504 match empty:
505 case True{}:
506 SDone{kind.val(kind, Nil{}), next, stack}
507 case False{}:
508 SHead{next, Con{Frame{kind, left, Nil{}}, stack}}
509
510# A map of n pairs has 2n items; an attribute adds the reply after its pairs.
511def agg.left(+kind: U32, +n: U32) -> U32:
512 Bool.pick(U32, U32.is_eq(kind, 1), (n * 2 : U32), Bool.pick(U32, U32.is_eq(kind, 3), (n * 2 + 1 : U32), n))
513
514def agg.bad(bad: Bool, +kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>) -> St:
515 match bad:
516 case True{}:
517 SBad{}
518 case False{}:
519 +left = agg.left(kind, n)
520 agg.open(U32.is_eq(left, 0), kind, left, next, stack)
521
522# *-1 is RESP2's null array.
523def agg.nil(nil: Bool, +kind: U32, +n: U32, +next: U32, stack: List<&1, Frame>) -> St:
524 match nil:
525 case True{}:
526 null.st(U32.is_eq(kind, 0), next, stack)
527 case False{}:
528 agg.bad(U32.is_eq(n, 4294967294), kind, n, next, stack)
529
530def agg.st(+kind: U32, +next: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Num) -> Bytes.Bytes & St:
531 (b, n) = r
532 +k = len.of(n)
533 (b, agg.nil(U32.is_eq(k, 4294967295), kind, k, next, stack))
534
535# The type byte t, then the line from s until the CRLF at e.
536def item(+t: U32, +s: U32, +e: U32, b: Bytes.Bytes, stack: List<&1, Frame>) -> Bytes.Bytes & St:
537 match t:
538 case 43:
539 line.st(0, (e + 2 : U32), stack, Bytes.slice(b, s, (e - s : U32)))
540 case 45:
541 line.st(1, (e + 2 : U32), stack, Bytes.slice(b, s, (e - s : U32)))
542 case 44:
543 line.st(2, (e + 2 : U32), stack, Bytes.slice(b, s, (e - s : U32)))
544 case 40:
545 line.st(3, (e + 2 : U32), stack, Bytes.slice(b, s, (e - s : U32)))
546 case 58:
547 int.st((e + 2 : U32), stack, num(b, s, e))
548 case 95:
549 (b, null.st(U32.is_eq(s, e), (e + 2 : U32), stack))
550 case 35:
551 boolean.st((e + 2 : U32), stack, U32.is_eq(e, (s + 1 : U32)), byte(b, s))
552 case 36:
553 blob.st(0, (e + 2 : U32), stack, num(b, s, e))
554 case 33:
555 blob.st(1, (e + 2 : U32), stack, num(b, s, e))
556 case 61:
557 blob.st(2, (e + 2 : U32), stack, num(b, s, e))
558 case 42:
559 agg.st(0, (e + 2 : U32), stack, num(b, s, e))
560 case 37:
561 agg.st(1, (e + 2 : U32), stack, num(b, s, e))
562 case 126:
563 agg.st(2, (e + 2 : U32), stack, num(b, s, e))
564 case 124:
565 agg.st(3, (e + 2 : U32), stack, num(b, s, e))
566 case _:
567 agg.st(4, (e + 2 : U32), stack, num(b, s, e))
568
569def head.eol(+t: U32, +i: U32, stack: List<&1, Frame>, r: Bytes.Bytes & Maybe<&2, U32>) -> Bytes.Bytes & St:
570 (b, m) = r
571 match m:
572 case Some{+e}:
573 item(t, (i + 1 : U32), e, b, stack)
574 case None{}:
575 (b, SMore{})
576
577def head.end(end: Bool) -> St:
578 match end:
579 case True{}:
580 SMore{}
581 case False{}:
582 SBad{}
583
584def head.known(ok: Bool, +t: U32, +i: U32, stack: List<&1, Frame>, b: Bytes.Bytes) -> Bytes.Bytes & St:
585 match ok:
586 case True{}:
587 head.eol(t, i, stack, Bytes.find.from(b, "\r\n", (i + 1 : U32)))
588 case False{}:
589 (b, head.end(U32.is_eq(t, 256)))
590
591def head(+i: U32, stack: List<&1, Frame>, r: Bytes.Bytes & U32) -> Bytes.Bytes & St:
592 (b, +t) = r
593 head.known(known(t), t, i, stack, b)
594
595def done.last(last: Bool, +kind: U32, +left: U32, xs: List<&1, Val>, +i: U32, rest: List<&1, Frame>) -> St:
596 match last:
597 case True{}:
598 SDone{kind.val(kind, List.reverse(&1, Val, xs)), i, rest}
599 case False{}:
600 SHead{i, Con{Frame{kind, (left - 1 : U32), xs}, rest}}
601
602def done(stack: List<&1, Frame>, v: Val, +i: U32) -> St:
603 match stack:
604 case Nil{}:
605 SOk{v, i}
606 case Con{fr, rest}:
607 Frame{+kind, +left, xs} = fr
608 done.last(U32.is_eq(left, 1), kind, left, Con{v, xs}, i, rest)
609
610def run(fuel: Nat, r: Bytes.Bytes & St) -> Bytes.Bytes & Step:
611 match fuel:
612 case 0n:
613 (b, st) = r
614 (b, Bad{})
615 case 1n+f:
616 (b, st) = r
617 match st:
618 case SOk{v, +i}:
619 (b, Reply{v, i})
620 case SMore{}:
621 (b, More{})
622 case SBad{}:
623 (b, Bad{})
624 case SHead{+i, stack}:
625 run(f, head(i, stack, byte(b, i)))
626 case SDone{v, +i, stack}:
627 run(f, (b, done(stack, v, i)))
628
629# One reply starting at byte i of b. Every item takes at least 3 bytes and each step
630# reads an item or closes one, so 2 * (len - i) + 2 steps are enough.
631def decode(b: Bytes.Bytes, +i: U32) -> Bytes.Bytes & Step:
632 Bytes.Bytes{+len, buf} = b
633 run(Nat.add(Nat.mul(U32.to_nat((len - i : U32)), 2n), 2n), (Bytes.Bytes{len, buf}, SHead{i, Nil{}}))
634
635# ---- incremental reader
636
637# Bytes read from the server, and the index of the first byte not yet decoded.
638type Reader is Type:
639 Reader{buf: Bytes.Bytes, pos: U32}
640
641def reader() -> Reader:
642 Reader{Bytes.new(0), 0}
643
644def feed.at(zero: Bool, buf: Bytes.Bytes, +pos: U32, chunk: Bytes.Bytes) -> Reader:
645 match zero:
646 case True{}:
647 Reader{Bytes.append(buf, chunk), 0}
648 case False{}:
649 Reader{Bytes.append(Bytes.snd(Bytes.slice(buf, pos, 4294967295)), chunk), 0}
650
651# Drops the decoded bytes, then adds chunk.
652# ponytail: a reply split over k reads is decoded from its start k times; keep state per item if huge arrays arrive slowly.
653def feed(r: Reader, chunk: Bytes.Bytes) -> Reader:
654 Reader{buf, +pos} = r
655 feed.at(U32.is_zero(pos), buf, pos, chunk)
656
657type Next is Type:
658 Got{v: Val}
659 Want{}
660 Broken{}
661
662def next.of(+pos: U32, r: Bytes.Bytes & Step) -> Reader & Next:
663 (b, s) = r
664 match s:
665 case Reply{v, +i}:
666 (Reader{b, i}, Got{v})
667 case More{}:
668 (Reader{b, pos}, Want{})
669 case Bad{}:
670 (Reader{b, pos}, Broken{})
671
672# The next whole reply, Want when it has not all arrived, or Broken for bytes that are not RESP.
673def next(r: Reader) -> Reader & Next:
674 Reader{buf, +pos} = r
675 next.of(pos, decode(buf, pos))
676
677def pending.of(+pos: U32, r: Bytes.Bytes & U32) -> Reader & U32:
678 (b, +len) = r
679 (Reader{b, pos}, (len - pos : U32))
680
681# The count of bytes read but not yet decoded.
682def pending(r: Reader) -> Reader & U32:
683 Reader{buf, +pos} = r
684 pending.of(pos, Bytes.length(buf))
685
686# ---- connections
687
688# host is a name or a numeric address; TLS checks the certificate against it. Empty user
689# and pass send no AUTH (a pass alone authenticates as "default"); db 0 sends no SELECT.
690# ms bounds each connect and read (0: none).
691type Config is Data:
692 Config{host: String, port: U32, tls: Bool, user: String, pass: String, db: U32, ms: U32}
693
694def config(+host: String, +port: U32) -> Config:
695 Config{host, port, False{}, "", "", 0, 10000}
696
697# A failed call closes its connection. Rejected carries the server's error reply to the handshake.
698type Failure is Type:
699 IoFail{code: U32, why: String}
700 Closed{}
701 Malformed{}
702 Rejected{reply: Val}
703
704# key names the pool bucket: TLS, user, host, port, and db.
705type Conn is Type:
706 Conn{tls: Bool, sock: Socket, rd: Reader, key: String, ms: U32}
707
708def io.close(tls: Bool, s: Socket) -> IO(Unit):
709 match tls:
710 case True{}:
711 Wire.tls.close(s)
712 case False{}:
713 Socket.close(s)
714
715def io.send(tls: Bool, s: Socket, +len: U32, buf: Array<U32>) -> IO(Socket & Result<&1, &1, U32 & String, Unit>):
716 match tls:
717 case True{}:
718 Wire.tls.send.words(s, len, buf)
719 case False{}:
720 Wire.send.words(s, len, buf)
721
722def io.recv(tls: Bool, s: Socket, +ms: U32) -> IO(Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>):
723 match tls:
724 case True{}:
725 Wire.tls.recv.words(s, 65536, ms)
726 case False{}:
727 Wire.recv.words(s, 65536, ms)
728
729def close(c: Conn) -> IO(Unit):
730 Conn{tls, sock, rd, key, ms} = c
731 io.close(tls, sock)
732
733def send.after(+tls: Bool, rd: Reader, +key: String, +ms: U32, m: Socket & Result<&1, &1, U32 & String, Unit>) -> IO(Result<&1, &1, Failure, Conn>):
734 (s, r) = m
735 match r:
736 case Fail{(code, why)}:
737 do IO<Result<&1, &1, Failure, Conn>>:
738 io.close(tls, s)
739 return Fail{IoFail{code, why}}
740 case Done{u}:
741 IO.pure(Result<&1, &1, Failure, Conn>, Done{Conn{tls, s, rd, key, ms}})
742
743def send(c: Conn, data: Bytes.Bytes) -> IO(Result<&1, &1, Failure, Conn>):
744 Conn{+tls, sock, rd, +key, +ms} = c
745 Bytes.Bytes{+len, buf} = data
746 do IO<Result<&1, &1, Failure, Conn>>:
747 m : Socket & Result<&1, &1, U32 & String, Unit> <- io.send(tls, sock, len, buf)
748 send.after(tls, rd, key, ms, m)
749
750# RNext holds the reader after its last decode attempt; RFail a socket to close.
751type Rd is Type:
752 RNext{tls: Bool, sock: Socket, key: String, ms: U32, r: Reader & Next}
753 RFail{tls: Bool, sock: Socket, why: Failure}
754
755def got.len(zero: Bool, +tls: Bool, +key: String, +ms: U32, rd: Reader, s: Socket, +len: U32, words: Array<U32>) -> Rd:
756 match zero:
757 case True{}:
758 RFail{tls, s, Closed{}}
759 case False{}:
760 RNext{tls, s, key, ms, next(feed(rd, Bytes.Bytes{len, words}))}
761
762def got(+tls: Bool, +key: String, +ms: U32, rd: Reader, m: Socket & Result<&1, &1, U32 & String, U32 & Array<U32>>) -> Rd:
763 (s, r) = m
764 match r:
765 case Fail{(code, why)}:
766 RFail{tls, s, IoFail{code, why}}
767 case Done{(+len, words)}:
768 got.len(U32.is_zero(len), tls, key, ms, rd, s, len, words)
769
770def read.fail(tls: Bool, sock: Socket, why: Failure) -> IO(Result<&1, &1, Failure, Conn & Val>):
771 do IO<Result<&1, &1, Failure, Conn & Val>>:
772 io.close(tls, sock)
773 return Fail{why}
774
775# One recv per step. 2^24 reads of up to 64 KiB pass Redis's 512 MiB bulk limit.
776def read.loop(fuel: Nat, st: Rd) -> IO(Result<&1, &1, Failure, Conn & Val>):
777 match fuel:
778 case 0n:
779 match st:
780 case RNext{tls, sock, key, ms, r}:
781 read.fail(tls, sock, Malformed{})
782 case RFail{tls, sock, why}:
783 read.fail(tls, sock, why)
784 case 1n+f:
785 match st:
786 case RFail{tls, sock, why}:
787 read.fail(tls, sock, why)
788 case RNext{+tls, sock, +key, +ms, r}:
789 (rd, nx) = r
790 match nx:
791 case Got{v}:
792 IO.pure(Result<&1, &1, Failure, Conn & Val>, Done{(Conn{tls, sock, rd, key, ms}, v)})
793 case Broken{}:
794 read.fail(tls, sock, Malformed{})
795 case Want{}:
796 do IO<Result<&1, &1, Failure, Conn & Val>>:
797 m : Socket & Result<&1, &1, U32 & String, U32 & Array<U32>> <- io.recv(tls, sock, ms)
798 read.loop(f, got(tls, key, ms, rd, m))
799
800# The next reply. Bytes already read are decoded before the socket is asked for more.
801def read(c: Conn) -> IO(Result<&1, &1, Failure, Conn & Val>):
802 Conn{tls, sock, rd, key, ms} = c
803 read.loop(Nat.mul(65536n, 256n), RNext{tls, sock, key, ms, next(rd)})
804
805def read.push(xs: List<&1, Val>, r: Result<&1, &1, Failure, Conn & Val>) -> Result<&1, &1, Failure, Conn & List<&1, Val>>:
806 match r:
807 case Fail{e}:
808 Fail{e}
809 case Done{(c, v)}:
810 Done{(c, Con{v, xs})}
811
812# n more replies onto the reversed list in st.
813def read.n(n: Nat, st: Result<&1, &1, Failure, Conn & List<&1, Val>>) -> IO(Result<&1, &1, Failure, Conn & List<&1, Val>>):
814 match n:
815 case 0n:
816 match st:
817 case Fail{e}:
818 IO.pure(Result<&1, &1, Failure, Conn & List<&1, Val>>, Fail{e})
819 case Done{(c, xs)}:
820 IO.pure(Result<&1, &1, Failure, Conn & List<&1, Val>>, Done{(c, List.reverse(&1, Val, xs))})
821 case 1n+p:
822 match st:
823 case Fail{e}:
824 IO.pure(Result<&1, &1, Failure, Conn & List<&1, Val>>, Fail{e})
825 case Done{(c, xs)}:
826 do IO<Result<&1, &1, Failure, Conn & List<&1, Val>>>:
827 r : Result<&1, &1, Failure, Conn & Val> <- read(c)
828 read.n(p, read.push(xs, r))
829
830def commands(cmds: List<&1, List<&1, Bytes.Bytes>>, o: Bytes.Bytes, +n: Nat) -> Bytes.Bytes & Nat:
831 match cmds:
832 case Nil{}:
833 (o, n)
834 case Con{h, t}:
835 commands(t, Bytes.append(o, encode.command(h)), (1n + n : Nat))
836
837def pipeline.sent(+n: Nat, r: Result<&1, &1, Failure, Conn>) -> IO(Result<&1, &1, Failure, Conn & List<&1, Val>>):
838 match r:
839 case Fail{e}:
840 IO.pure(Result<&1, &1, Failure, Conn & List<&1, Val>>, Fail{e})
841 case Done{c}:
842 read.n(n, Done{(c, Nil{})})
843
844def pipeline.of(c: Conn, r: Bytes.Bytes & Nat) -> IO(Result<&1, &1, Failure, Conn & List<&1, Val>>):
845 (data, +n) = r
846 do IO<Result<&1, &1, Failure, Conn & List<&1, Val>>>:
847 s : Result<&1, &1, Failure, Conn> <- send(c, data)
848 pipeline.sent(n, s)
849
850# Sends every command in one write, then reads one reply per command, in order.
851# Error replies come back as Error or BulkError values, not as failures.
852def pipeline(c: Conn, cmds: List<&1, List<&1, Bytes.Bytes>>) -> IO(Result<&1, &1, Failure, Conn & List<&1, Val>>):
853 pipeline.of(c, commands(cmds, Bytes.new(0), 0n))
854
855def command.sent(r: Result<&1, &1, Failure, Conn>) -> IO(Result<&1, &1, Failure, Conn & Val>):
856 match r:
857 case Fail{e}:
858 IO.pure(Result<&1, &1, Failure, Conn & Val>, Fail{e})
859 case Done{c}:
860 read(c)
861
862# One command and its reply, for example command(c, args(["SET", "k", "v"])).
863def command(c: Conn, argv: List<&1, Bytes.Bytes>) -> IO(Result<&1, &1, Failure, Conn & Val>):
864 do IO<Result<&1, &1, Failure, Conn & Val>>:
865 r : Result<&1, &1, Failure, Conn> <- send(c, encode.command(argv))
866 command.sent(r)
867
868def get(c: Conn, key: Bytes.Bytes) -> IO(Result<&1, &1, Failure, Conn & Val>):
869 command(c, [text("GET"), key])
870
871def set(c: Conn, key: Bytes.Bytes, value: Bytes.Bytes) -> IO(Result<&1, &1, Failure, Conn & Val>):
872 command(c, [text("SET"), key, value])
873
874def incr(c: Conn, key: Bytes.Bytes) -> IO(Result<&1, &1, Failure, Conn & Val>):
875 command(c, [text("INCR"), key])
876
877def first.err(xs: List<&1, Val>) -> Maybe<&1, Val>:
878 match xs:
879 case Nil{}:
880 None{}
881 case Con{Error{t}, rest}:
882 Some{Error{t}}
883 case Con{BulkError{t}, rest}:
884 Some{BulkError{t}}
885 case Con{v, rest}:
886 first.err(rest)
887
888def hello.check(c: Conn, e: Maybe<&1, Val>) -> IO(Result<&1, &1, Failure, Conn>):
889 match e:
890 case None{}:
891 IO.pure(Result<&1, &1, Failure, Conn>, Done{c})
892 case Some{v}:
893 do IO<Result<&1, &1, Failure, Conn>>:
894 close(c)
895 return Fail{Rejected{v}}
896
897def hello.got(r: Result<&1, &1, Failure, Conn & List<&1, Val>>) -> IO(Result<&1, &1, Failure, Conn>):
898 match r:
899 case Fail{e}:
900 IO.pure(Result<&1, &1, Failure, Conn>, Fail{e})
901 case Done{(c, xs)}:
902 hello.check(c, first.err(xs))
903
904def hello.auth(+user: String, +pass: String) -> List<&2, String>:
905 Bool.pick(List<&2, String>, String.eq(pass, ""), Nil{}, ["AUTH", Bool.pick(String, String.eq(user, ""), "default", user), pass])
906
907def hello.select(+db: U32) -> List<&1, List<&1, Bytes.Bytes>>:
908 Bool.pick(List<&1, List<&1, Bytes.Bytes>>, U32.is_zero(db), Nil{}, [args(["SELECT", U32.show(db)])])
909
910# HELLO 3 (with AUTH), then SELECT, in one pipeline.
911def hello(c: Conn, +user: String, +pass: String, +db: U32) -> IO(Result<&1, &1, Failure, Conn>):
912 do IO<Result<&1, &1, Failure, Conn>>:
913 r : Result<&1, &1, Failure, Conn & List<&1, Val>> <- pipeline(c, Con{args(List.append(&2, String, ["HELLO", "3"], hello.auth(user, pass))), hello.select(db)})
914 hello.got(r)
915
916def key(cfg: Config) -> String:
917 Config{+host, +port, +tls, +user, pass, +db, ms} = cfg
918 Bool.pick(String, tls, "rediss://", "redis://") ++ user ++ "@" ++ host ++ ":" ++ U32.show(port) ++ "/" ++ U32.show(db)
919
920def tls.done(+key: String, +ms: U32, +user: String, +pass: String, +db: U32, m: Socket & Result<&1, &1, U32 & String, Unit>) -> IO(Result<&1, &1, Failure, Conn>):
921 (s, r) = m
922 match r:
923 case Fail{(code, why)}:
924 do IO<Result<&1, &1, Failure, Conn>>:
925 Wire.tls.close(s)
926 return Fail{IoFail{code, why}}
927 case Done{u}:
928 hello(Conn{True{}, s, reader(), key, ms}, user, pass, db)
929
930def dialed(r: Result<&1, &1, U32 & String, Socket>, tls: Bool, +k: String, +host: String, +user: String, +pass: String, +db: U32, +ms: U32) -> IO(Result<&1, &1, Failure, Conn>):
931 match r:
932 case Fail{(code, why)}:
933 IO.pure(Result<&1, &1, Failure, Conn>, Fail{IoFail{code, why}})
934 case Done{s}:
935 match tls:
936 case True{}:
937 do IO<Result<&1, &1, Failure, Conn>>:
938 m : Socket & Result<&1, &1, U32 & String, Unit> <- Wire.tls.connect(s, host, ms)
939 tls.done(k, ms, user, pass, db, m)
940 case False{}:
941 hello(Conn{False{}, s, reader(), k, ms}, user, pass, db)
942
943def dial.after(r: Result<&1, &1, U32 & String, Socket>, +ip: String, +port: U32, +ms: U32) -> IO(Result<&1, &1, U32 & String, Socket>):
944 match r:
945 case Done{s}:
946 IO.pure(Result<&1, &1, U32 & String, Socket>, Done{s})
947 case Fail{e}:
948 Wire.connect(ip, port, ms)
949
950def dial.more(prev: IO(Result<&1, &1, U32 & String, Socket>), +ip: String, +port: U32, +ms: U32) -> IO(Result<&1, &1, U32 & String, Socket>):
951 do IO<Result<&1, &1, U32 & String, Socket>>:
952 r : Result<&1, &1, U32 & String, Socket> <- prev
953 dial.after(r, ip, port, ms)
954
955# Each address in turn, until one connects.
956def dial(ips: List<&2, String>, +port: U32, +ms: U32, acc: IO(Result<&1, &1, U32 & String, Socket>)) -> IO(Result<&1, &1, U32 & String, Socket>):
957 match ips:
958 case Nil{}:
959 acc
960 case Con{ip, t}:
961 dial(t, port, ms, dial.more(acc, ip, port, ms))
962
963def connect.to(+k: String, cfg: Config) -> IO(Result<&1, &1, Failure, Conn>):
964 Config{+host, +port, +tls, +user, +pass, +db, +ms} = cfg
965 do IO<Result<&1, &1, Failure, Conn>>:
966 ips : List<&2, String> <- Dns.resolve.all(host)
967 r : Result<&1, &1, U32 & String, Socket> <- dial(ips, port, ms, IO.pure(Result<&1, &1, U32 & String, Socket>, Fail{(0, "no address for " ++ host)}))
968 dialed(r, tls, k, host, user, pass, db, ms)
969
970# TCP (and TLS, which checks the certificate and host name), then the handshake.
971def connect(+cfg: Config) -> IO(Result<&1, &1, Failure, Conn>):
972 connect.to(key(cfg), cfg)
973
974# ---- pool
975
976# Idle connections per key, up to cap per key.
977type Pool is Type:
978 Pool{cap: U32, idle: Map<&1, List<&1, Conn>>}
979
980def pool.new.with(+cap: U32) -> Pool:
981 Pool{cap, Map.new(&1, List<&1, Conn>)}
982
983def pool.new() -> Pool:
984 pool.new.with(8)
985
986def pool.take(+cap: U32, +k: String, r: Map<&1, List<&1, Conn>> & Maybe<&1, List<&1, Conn>>) -> Pool & Maybe<&1, Conn>:
987 (m, got) = r
988 match got:
989 case None{}:
990 (Pool{cap, m}, None{})
991 case Some{xs}:
992 match xs:
993 case Nil{}:
994 (Pool{cap, m}, None{})
995 case Con{c, t}:
996 (Pool{cap, Map.set(&1, List<&1, Conn>, m, k, t)}, Some{c})
997
998def pool.opened(p: Pool, r: Result<&1, &1, Failure, Conn>) -> IO(Pool & Result<&1, &1, Failure, Conn>):
999 IO.pure(Pool & Result<&1, &1, Failure, Conn>, (p, r))
1000
1001def pool.got(cfg: Config, r: Pool & Maybe<&1, Conn>) -> IO(Pool & Result<&1, &1, Failure, Conn>):
1002 (p, m) = r
1003 match m:
1004 case Some{c}:
1005 IO.pure(Pool & Result<&1, &1, Failure, Conn>, (p, Done{c}))
1006 case None{}:
1007 do IO<Pool & Result<&1, &1, Failure, Conn>>:
1008 c : Result<&1, &1, Failure, Conn> <- connect(cfg)
1009 pool.opened(p, c)
1010
1011# The idle connection given back last, or a new one.
1012# ponytail: an idle connection the server has since closed fails on its first command; add a PING check if that bites.
1013def pool.get(p: Pool, +cfg: Config) -> IO(Pool & Result<&1, &1, Failure, Conn>):
1014 Pool{+cap, idle} = p
1015 +k = key(cfg)
1016 pool.got(cfg, pool.take(cap, k, Map.pop(&1, List<&1, Conn>, idle, k)))
1017
1018# The first n connections; the rest are closed.
1019def conns.cut(xs: List<&1, Conn>, n: Nat) -> IO(List<&1, Conn>):
1020 match xs:
1021 case Nil{}:
1022 IO.pure(List<&1, Conn>, Nil{})
1023 case Con{c, t}:
1024 match n:
1025 case 0n:
1026 do IO<List<&1, Conn>>:
1027 close(c)
1028 conns.cut(t, 0n)
1029 case 1n+k:
1030 do IO<List<&1, Conn>>:
1031 r : List<&1, Conn> <- conns.cut(t, k)
1032 return c <> r
1033
1034def pool.keep(+cap: U32, +k: String, c: Conn, r: Map<&1, List<&1, Conn>> & Maybe<&1, List<&1, Conn>>) -> IO(Pool):
1035 (m, old) = r
1036 do IO<Pool>:
1037 xs : List<&1, Conn> <- conns.cut(c <> Maybe.default(&1, List<&1, Conn>, old, Nil{}), U32.to_nat(cap))
1038 return Pool{cap, Map.set(&1, List<&1, Conn>, m, k, xs)}
1039
1040def pool.give(p: Pool, c: Conn) -> IO(Pool):
1041 Pool{+cap, idle} = p
1042 Conn{tls, sock, rd, +k, ms} = c
1043 pool.keep(cap, k, Conn{tls, sock, rd, k, ms}, Map.pop(&1, List<&1, Conn>, idle, k))
1044
1045def pool.clean(clean: Bool, p: Pool, c: Conn) -> IO(Pool):
1046 match clean:
1047 case True{}:
1048 pool.give(p, c)
1049 case False{}:
1050 do IO<Pool>:
1051 close(c)
1052 return p
1053
1054def pool.left(p: Pool, +tls: Bool, sock: Socket, +key: String, +ms: U32, r: Reader & U32) -> IO(Pool):
1055 (rd, +n) = r
1056 pool.clean(U32.is_zero(n), p, Conn{tls, sock, rd, key, ms})
1057
1058# Gives c back for reuse. A connection with unread reply bytes is out of step, so it is closed.
1059def pool.put(p: Pool, c: Conn) -> IO(Pool):
1060 Conn{+tls, sock, rd, +key, +ms} = c
1061 pool.left(p, tls, sock, key, ms, pending(rd))
1062
1063def pool.close.go(xs: List<&1, List<&1, Conn>>) -> IO(Unit):
1064 match xs:
1065 case Nil{}:
1066 IO.pure(Unit, Unit{})
1067 case Con{cs, t}:
1068 do IO<Unit>:
1069 none : List<&1, Conn> <- conns.cut(cs, 0n)
1070 pool.close.go(t)
1071
1072def pool.close(p: Pool) -> IO(Unit):
1073 Pool{cap, idle} = p
1074 pool.close.go(Map.values(&1, List<&1, Conn>, idle))