From bf7c9842f4262a46b2ac120e1e2e4f8814f17f35 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 9 Aug 2026 14:08:54 +0000 Subject: [PATCH 1/4] fix: parse redis replies through a cursor instead of re-copying the buffer std.redis buffered every reply by appending to an immutable string, so each 4 KiB read copied everything read so far. The reply size a server may advertise is bounded at MAX_BULK_LEN, 512 MiB, which is about 131,000 appends averaging 256 MiB each: a bulk reply large enough is not slow, it never returns. Every RESP reply went through that one path. The buffer is now bytes plus a read cursor. Consuming a line or a payload moves the cursor, and the consumed prefix is dropped in one step when the next read arrives, so no byte is copied twice on the way through. Line scanning resumes where the previous scan stopped, one byte back so a CR at a read boundary still pairs with its LF. A bulk read knows the length up front, so it reads the remainder straight into a growing buffer and asks the socket for exactly what is left, which also means nothing is read past the payload. Reading bytes rather than a lossily-decoded string fixes two things on the way. A payload containing a NUL byte used to shorten the buffer the client measured its progress against, so the client waited for bytes that had already arrived; that stream now stays in sync. And bytes that are not utf-8 surface as an error instead of a string silently filled with replacement characters. The up-front reservation for a bulk read is capped, so a twelve-byte header advertising 512 MiB cannot turn into a 512 MiB allocation before any payload byte has arrived. --- std/redis.pith | 123 +++++++++++++++--- tests/cases/test_redis_split_reads.pith | 151 ++++++++++++++++++++++ tests/expected/test_redis_split_reads.txt | 22 ++++ 3 files changed, 280 insertions(+), 16 deletions(-) create mode 100644 tests/cases/test_redis_split_reads.pith create mode 100644 tests/expected/test_redis_split_reads.txt diff --git a/std/redis.pith b/std/redis.pith index 016d2580..f3cc3653 100644 --- a/std/redis.pith +++ b/std/redis.pith @@ -5,9 +5,15 @@ # a tagged value that mirrors RESP: a status line, an error, an integer, a bulk # string, nil, or an array of nested replies. # +# a reply is text: a bulk string carrying bytes that are not utf-8 comes back +# as an error rather than as a string quietly filled with replacement +# characters. reach for a redis client that speaks bytes if you store blobs. +# # i/o goes straight to the tcp_* builtins, the same ones std.io wraps; the # std.net.tcp module is a thin alias over them and isn't needed here. +import std.bytes as bytes + # replies come off an untrusted socket, so the length and count fields a server # advertises are bounded before we act on them: a hostile or buggy peer must not # be able to drive the client to OOM or overflow the native stack. @@ -16,6 +22,19 @@ MAX_BULK_LEN := 536870912 # 512 MiB — a single bulk string MAX_ARRAY_ITEMS := 10485760 # elements in one array reply MAX_REPLY_DEPTH := 64 # nested-array nesting +# bytes asked for in one socket read. +READ_CHUNK := 65536 + +# the most a bulk read reserves up front. the payload length comes off the +# wire, so reserving all of it would let a twelve-byte header ask for half a +# gigabyte before a single payload byte has arrived; the buffer grows as the +# data actually lands instead. +MAX_RESERVE := 262144 + +# the two bytes that terminate every RESP line. +CR := 13 +LF := 10 + # a reply from redis. `Bulk` and `Array` may be absent, reported as `Nil`. # an array's elements are flattened to strings (a scalar becomes its text, a # nil becomes ""), which covers the common list-returning commands; reach for @@ -44,48 +63,120 @@ fn reply_is_error(r: Reply) -> Bool: Reply.Error(_) => true _ => false -# a connection to a redis server. `buf` holds bytes read past the current reply. +# a connection to a redis server. `buf` holds the bytes read off the socket +# that have not been parsed yet, and `pos` is how far into them the parser has +# got. the cursor is what keeps the client linear: consuming a line or a +# payload only moves `pos`, and the bytes already consumed are dropped in one +# step when the next read comes in, so no byte of a reply is copied twice. pub struct Client: fd: Int - mut buf: String + mut buf: Bytes + mut pos: Int # open a connection to a redis server. pub fn connect(host: String, port: Int) -> Client!: fd := tcp_connect(host, port)! - return Client(fd: fd, buf: "") + return Client(fd: fd, buf: bytes.empty(), pos: 0) impl Client: - # pull one more chunk from the socket into the buffer; false at end of stream. + # bytes read off the socket that the parser has not consumed yet. + fn buffered() -> Int: + return self.buf.len() - self.pos + + # decode a byte range of the buffer as text. a reply is text as far as this + # client is concerned, so bytes that are not utf-8 are an error rather than + # a string quietly filled with replacement characters. + fn text(start: Int, end: Int) -> String!: + decoded := bytes.substring_utf8(self.buf, start, end) + if decoded.is_err: + fail "redis: reply is not valid utf-8" + return decoded.ok + + # pull one more chunk from the socket into the buffer; false at end of + # stream. the consumed prefix is dropped here, once per read, rather than + # on every parse step. fn fill() -> Bool!: - chunk := tcp_read(self.fd, 4096)! + chunk := tcp_read_bytes(self.fd, READ_CHUNK)! if chunk.len() == 0: return false - self.buf = self.buf + chunk + if self.buffered() == 0: + self.buf = chunk + else: + self.buf = self.buf.slice(self.pos, self.buf.len()).concat(chunk) + self.pos = 0 return true + # index of the first CRLF at or after `start`, or -1 when there is none in + # the bytes read so far. + fn find_crlf(start: Int) -> Int: + mut i := start + if i < self.pos: + i = self.pos + limit := self.buf.len() - 1 + while i < limit: + if self.buf[i] == CR and self.buf[i + 1] == LF: + return i + i = i + 1 + return -1 + # read one CRLF-terminated line, returning it without the terminator. fn read_line() -> String!: - mut idx := self.buf.index_of("\r\n") + mut idx := self.find_crlf(self.pos) while idx < 0: # a server that never sends the terminator must not grow the buffer # without bound. - if self.buf.len() > MAX_LINE_LEN: + if self.buffered() > MAX_LINE_LEN: fail "redis: reply line exceeds maximum length" + searched := self.buffered() if not self.fill()!: fail "redis: connection closed" - idx = self.buf.index_of("\r\n") - line := self.buf.substring(0, idx) - self.buf = self.buf.substring(idx + 2, self.buf.len()) + # resume one byte back, so a CR that was the last byte of the + # previous read still pairs with the LF that follows it. every + # other byte is looked at once, whatever sizes the reply arrives in. + mut resume := self.pos + searched - 1 + if resume < self.pos: + resume = self.pos + idx = self.find_crlf(resume) + line := self.text(self.pos, idx)! + self.pos = idx + 2 return line # read exactly `count` payload bytes plus the trailing CRLF. fn read_payload(count: Int) -> String!: - while self.buf.len() < count + 2: - if not self.fill()!: + need := count + 2 + if self.buffered() >= need: + data := self.text(self.pos, self.pos + count)! + self.pos = self.pos + need + return data + # the length is known, so the rest of the payload is read straight into + # a buffer that grows: the payload is copied once on the way in and + # once on the way out, however many reads it takes to arrive. + mut reserve := need + if reserve > MAX_RESERVE: + reserve = MAX_RESERVE + out := bytes.buffer_with_capacity(reserve) + errdefer out.free() + mut got := self.buffered() + if got > 0: + out.write_range(self.buf, self.pos, got) + self.buf = bytes.empty() + self.pos = 0 + while got < need: + mut want := need - got + if want > READ_CHUNK: + want = READ_CHUNK + chunk := tcp_read_bytes(self.fd, want)! + if chunk.len() == 0: fail "redis: connection closed" - data := self.buf.substring(0, count) - self.buf = self.buf.substring(count + 2, self.buf.len()) - return data + out.write(chunk)! + got = got + chunk.len() + # nothing is read past the payload, so the buffer is empty afterwards + # and the gathered bytes are released as soon as the text is out. + gathered := out.take_bytes() + payload := bytes.substring_utf8(gathered, 0, count) + if payload.is_err: + fail "redis: reply is not valid utf-8" + return payload.ok # parse one reply off the wire, recursing for arrays. fn read_reply() -> Reply!: diff --git a/tests/cases/test_redis_split_reads.pith b/tests/cases/test_redis_split_reads.pith new file mode 100644 index 00000000..98c8e39f --- /dev/null +++ b/tests/cases/test_redis_split_reads.pith @@ -0,0 +1,151 @@ +# a redis reply must parse the same however the socket happens to break it up. +# +# this drives the client with one RESP script delivered whole, then a byte at a +# time, then in odd-sized pieces, and compares what came back each time. the +# byte-at-a-time run is the interesting one: it splits every CRLF across two +# reads, which is what the parser's scan resume has to get right, and it puts a +# read boundary at every offset inside every reply. the script covers the reply +# shapes — status, error, integer, bulk (including empty and null), flat and +# nested arrays — and a bulk large enough to arrive in many reads. the limits +# that must still reject get their own connection each, since a rejected reply +# leaves the stream unparsed. + +import std.redis as redis +from std.redis import Reply +import std.time as time +from std.collections import copy_list + +fn min_int(a: Int, b: Int) -> Int: + if a < b: + return a + return b + +fn repeat(text: String, times: Int) -> String: + mut out := "" + mut i := 0 + while i < times: + out = out + text + i = i + 1 + return out + +# answer each command with the next canned reply, written `piece` bytes at a +# time. the pause between pieces is what keeps the kernel from coalescing them +# back into one read — without it the client would see the whole reply at once +# and the split would never be exercised. `piece` at or above the reply length +# means one write, and no pauses. +fn serve(listener: Int, script: List[String], piece: Int) -> Int!: + conn := tcp_accept(listener)! + for reply in script: + request := tcp_read(conn, 4096)! + if request.len() == 0: + tcp_close(conn) + return 0 + mut off := 0 + while off < reply.len(): + stop := min_int(off + piece, reply.len()) + wrote := tcp_write(conn, reply.substring(off, stop)) catch 0 + if wrote == 0: + tcp_close(conn) + return 0 + off = off + wrote + if piece < reply.len(): + time.delay(1) + tcp_close(conn) + return 0 + +# a stable rendering of a reply, so two runs can be compared as text. +fn render(reply: Reply) -> String: + return match reply: + Reply.Status(s) => "status:" + s + Reply.Error(s) => "error:" + s + Reply.Integer(n) => "int:" + n.to_string() + Reply.Bulk(s) => "bulk:" + s.len().to_string() + ":" + s + Reply.Nil => "nil" + Reply.Array(items) => "array:" + items.len().to_string() + ":" + items.join("|") + +# run the whole script against a fresh stub on `port` and return one rendering +# per reply. a command that fails renders its error, so a `-ERR` reply and a +# rejected limit are compared as text like everything else. +fn run_script(port: Int, script: List[String], piece: Int) -> List[String]!: + listener := tcp_listen("127.0.0.1", port)! + # the stub reads the script on another task, so it gets its own copy. + task := spawn serve(listener, copy_list(script), piece) + client := redis.connect("127.0.0.1", port)! + mut out: List[String] := [] + mut i := 0 + while i < script.len(): + reply := client.command(["PROBE", i.to_string()]) + if reply.is_err: + out.push("failed:" + reply.err) + else: + out.push(render(reply.ok)) + i = i + 1 + client.close() + await task + tcp_close(listener) + return out + +# the reply shapes, small enough that a byte-at-a-time run stays quick. +fn shapes_script() -> List[String]: + return [ + "+PONG\r\n", + "-ERR unknown command\r\n", + ":42\r\n", + ":-1\r\n", + "$5\r\nhello\r\n", + "$0\r\n\r\n", + "$-1\r\n", + "*0\r\n", + "*-1\r\n", + "*3\r\n$3\r\nfoo\r\n:7\r\n$-1\r\n", + "*2\r\n*2\r\n$1\r\na\r\n$1\r\nb\r\n$1\r\nc\r\n", + "$" + repeat("x", 300).len().to_string() + "\r\n" + repeat("x", 300) + "\r\n", + ] + +# one bulk reply far larger than a single read, to exercise the gather path. +fn bulk_script() -> List[String]: + payload := repeat(repeat("abcdefgh", 128), 256) + return ["$" + payload.len().to_string() + "\r\n" + payload + "\r\n"] + +# a rendering short enough to print. the comparison always uses the full text; +# only what lands in the expected output is shortened. +fn brief(text: String) -> String: + if text.len() <= 40: + return text + return text.substring(0, 40) + "...(" + text.len().to_string() + ")" + +fn compare(label: String, expected: List[String], actual: List[String]): + if expected.len() != actual.len(): + print(label + ": MISMATCH (" + expected.len().to_string() + " vs " + actual.len().to_string() + " replies)") + return + mut i := 0 + while i < expected.len(): + if expected[i] != actual[i]: + print(label + ": MISMATCH at " + i.to_string() + ": " + expected[i] + " != " + actual[i]) + return + i = i + 1 + print(label + ": same") + +fn main() -> Int!: + shapes := shapes_script() + whole := run_script(17510, shapes, 65536)! + for line in whole: + print(brief(line)) + + compare("1-byte pieces", whole, run_script(17511, shapes, 1)!) + compare("3-byte pieces", whole, run_script(17512, shapes, 3)!) + compare("7-byte pieces", whole, run_script(17513, shapes, 7)!) + + big := bulk_script() + big_whole := run_script(17514, big, 1048576)! + print(brief(big_whole[0])) + compare("262144-byte bulk in 997-byte pieces", big_whole, run_script(17515, big, 997)!) + compare("262144-byte bulk in 4096-byte pieces", big_whole, run_script(17516, big, 4096)!) + + # the limits get a connection each: a rejected reply leaves the rest of the + # stream unparsed, so nothing can follow it on the same socket. + print(run_script(17517, ["$536870913\r\n"], 65536)![0]) + print(run_script(17518, ["*10485761\r\n"], 65536)![0]) + print(run_script(17519, [repeat("*1\r\n", 66) + ":1\r\n"], 65536)![0]) + print(run_script(17520, [repeat("x", 70000)], 65536)![0]) + return 0 diff --git a/tests/expected/test_redis_split_reads.txt b/tests/expected/test_redis_split_reads.txt new file mode 100644 index 00000000..52757618 --- /dev/null +++ b/tests/expected/test_redis_split_reads.txt @@ -0,0 +1,22 @@ +status:PONG +failed:redis: ERR unknown command +int:42 +int:-1 +bulk:5:hello +bulk:0: +nil +array:0: +nil +array:3:foo|7| +array:2:|c +bulk:300:xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx...(309) +1-byte pieces: same +3-byte pieces: same +7-byte pieces: same +bulk:262144:abcdefghabcdefghabcdefghabcd...(262156) +262144-byte bulk in 997-byte pieces: same +262144-byte bulk in 4096-byte pieces: same +failed:redis: bulk string exceeds maximum length +failed:redis: array exceeds maximum length +failed:redis: reply nesting too deep +failed:redis: reply line exceeds maximum length From 148ffb81fadd906c4a69df4c66e4d0cc4213689a Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 9 Aug 2026 14:30:52 +0000 Subject: [PATCH 2/4] perf: build the http/2 request head through a buffer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The http/2 server bridges a decoded header list into an http/1.1 request head, and it did that with `head = head + name + ": " + value + crlf` — four appends per field, each copying everything written before it. open_stream has already capped the list at MAX_HEADER_COUNT fields and MAX_HEADER_LIST_BYTES, so the worst request that reaches the bridge is bounded, and that bound is what keeps this from being an availability bug. It still means about twelve megabytes of copying to produce a head of sixty kilobytes, and it is paid on every request. The head now goes into a byte buffer, which build_http_request wanted as bytes anyway, so the final string-to-bytes conversion goes with it. Appending an empty string to a ByteBuffer was an error rather than a no-op: write_string_utf8 returns the number of bytes written and the result convention reads zero as failure. ByteBuffer.write already short-circuits an empty write for exactly this reason and write_string_utf8 did not, which made a header sent with an empty value — legal, and common enough — drop the whole request. It now short-circuits the same way, with a test on each side of the boundary. --- std/bytes.pith | 14 +++ std/net/http2/server.pith | 23 ++++- .../cases/test_http2_empty_header_value.pith | 99 +++++++++++++++++++ .../test_http2_empty_header_value.txt | 1 + 4 files changed, 132 insertions(+), 5 deletions(-) create mode 100644 tests/cases/test_http2_empty_header_value.pith create mode 100644 tests/expected/test_http2_empty_header_value.txt diff --git a/std/bytes.pith b/std/bytes.pith index 373a30c6..9142fb31 100644 --- a/std/bytes.pith +++ b/std/bytes.pith @@ -136,6 +136,13 @@ impl ByteBuffer: return byte_buffer_write_byte(self.handle, value) fn write_string_utf8(text: String) -> Int!: + # same reasoning as write: the builtin reports bytes written and the + # result convention reads 0 as failure, so appending an empty string + # has to short-circuit rather than report an error for a no-op. text + # assembled piece by piece — a header value, a field that happens to + # be blank — is full of empty pieces. + if text.len() == 0: + return 0 return byte_buffer_write_string_utf8(self.handle, text) fn write_line_utf8(text: String) -> Int!: @@ -257,6 +264,13 @@ test "byte buffers write and reset": out.reset() assert_eq(out.bytes().len(), 0) + # an empty append is a no-op, not an error — the same contract write() has + empty_write := out.write_string_utf8("") + assert(not empty_write.is_err) + assert_eq(empty_write.ok, 0) + assert_eq(out.write_string_utf8("ok")!, 2) + assert_eq(out.bytes().to_string_utf8()!, "ok") + test "byte buffers append ranges and packed words": src := from_string_utf8("abcdef") out := buffer() diff --git a/std/net/http2/server.pith b/std/net/http2/server.pith index c2dc8513..68c85009 100644 --- a/std/net/http2/server.pith +++ b/std/net/http2/server.pith @@ -638,15 +638,28 @@ fn build_request(fields: List[hpack.HeaderField], body: Bytes) -> http.HttpReque path := pseudo(fields, ":path") authority := pseudo(fields, ":authority") crlf := "\r\n" - mut head := method + " " + path + " HTTP/1.1" + crlf - head = head + "host: " + authority + crlf + # the head is appended to a byte buffer rather than to a string. pith + # strings are immutable, so `head = head + field` re-copies everything + # written so far, four times per header field. open_stream has already + # capped the list at MAX_HEADER_COUNT fields and MAX_HEADER_LIST_BYTES, so + # the worst request that reaches here is bounded — but the bound still means + # about twelve megabytes of copying for a head of sixty kilobytes, and this + # runs once per request. a buffer appends each piece once, and + # build_http_request wants bytes anyway. + head := bytes.buffer() + errdefer head.free() + head.write_string_utf8(method + " " + path + " HTTP/1.1" + crlf)! + head.write_string_utf8("host: " + authority + crlf)! for field in fields: # pseudo-headers are h2-only; they became the request line and Host above. if field.name.starts_with(":"): continue - head = head + field.name + ": " + field.value + crlf - head = head + crlf - return http.build_http_request(bytes.from_string_utf8(head), body)! + head.write_string_utf8(field.name)! + head.write_string_utf8(": ")! + head.write_string_utf8(field.value)! + head.write_string_utf8(crlf)! + head.write_string_utf8(crlf)! + return http.build_http_request(head.take_bytes(), body)! # --- the response bridge --- diff --git a/tests/cases/test_http2_empty_header_value.pith b/tests/cases/test_http2_empty_header_value.pith new file mode 100644 index 00000000..33c9e7e5 --- /dev/null +++ b/tests/cases/test_http2_empty_header_value.pith @@ -0,0 +1,99 @@ +# a header sent with an empty value must survive the http/2 request bridge. +# +# the bridge synthesizes an http/1.1 request head from the decoded header list, +# and it builds that head through a byte buffer. an empty value is an empty +# append, and an append that reports "wrote nothing" reads as a failure under +# the result convention — so an empty value is exactly the case that turns a +# legal request into a dropped one. this drives a real h2c request carrying +# one empty-valued header, one ordinary header, and one whose value is empty +# between two that are not, and checks the server saw all three. + +import std.net.http2.server as server +import std.net.http2.frames as frames +import std.net.http2.hpack as hpack +import std.net.http as http +import std.bytes as bytes +import std.binary as binary +import std.encoding as encoding + +# report the head the bridge synthesized, with the line breaks made visible. +# the raw head is what proves an empty value came through: the accessor map +# cannot tell a header with an empty value from one that was never sent. +fn echo_headers(req: http.HttpRequestBytes) -> http.HttpResponse: + head := req.headers() catch "" + return http.text(200, head.replace("\r\n", "|")) + +fn serve_once(listener: Int): + client_fd := tcp_accept(listener) catch 0 + if client_fd != 0: + server.serve_h2c_connection(client_fd, echo_headers) + +fn send_request(fd: Int): + preface := encoding.from_hex("505249202a20485454502f322e300d0a0d0a534d0d0a0d0a") catch bytes.empty() + tcp_write_bytes(fd, preface) catch 0 + settings := frames.frame_bytes(frames.settings_frame([]) catch frames.settings_ack_frame()) catch bytes.empty() + tcp_write_bytes(fd, settings) catch 0 + encoder := hpack.new_encoder(hpack.DEFAULT_HEADER_TABLE_SIZE, false) + block := encoder.encode([ + hpack.header_field(":method", "GET"), + hpack.header_field(":path", "/hi"), + hpack.header_field(":scheme", "http"), + hpack.header_field(":authority", "localhost"), + hpack.header_field("x-empty", ""), + hpack.header_field("x-filled", "value"), + hpack.header_field("x-trailing", ""), + ]) + headers := frames.frame_bytes(frames.headers_frame(1, block, true, true)) catch bytes.empty() + tcp_write_bytes(fd, headers) catch 0 + +# read frames until a DATA frame ends stream 1, collecting the body bytes. +fn read_body(fd: Int) -> Bytes: + body := bytes.buffer() + mut buf := bytes.empty() + mut pos := 0 + mut rounds := 0 + while rounds < 200: + rounds = rounds + 1 + while buf.len() - pos < 9: + chunk := tcp_read_bytes(fd, 4096) catch bytes.empty() + if chunk.len() == 0: + out := body.bytes() + body.free() + return out + buf = bytes.concat(buf.slice(pos, buf.len()), chunk) + pos = 0 + length := binary.read_u24_be_at(buf, pos).unwrap_or(0) + total := 9 + length + while buf.len() - pos < total: + chunk := tcp_read_bytes(fd, 4096) catch bytes.empty() + if chunk.len() == 0: + out := body.bytes() + body.free() + return out + buf = bytes.concat(buf.slice(pos, buf.len()), chunk) + pos = 0 + full := buf.slice(pos, pos + total) + pos = pos + total + frame := frames.read_frame(binary.reader(full), frames.MAX_ALLOWED_FRAME_SIZE) catch frames.settings_ack_frame() + if frame.header.frame_type == frames.FRAME_DATA: + df := frames.parse_data(frame) catch frames.DataFrame(0, bytes.empty(), true) + body.write(df.data) catch 0 + if df.end_stream: + out := body.bytes() + body.free() + return out + out := body.bytes() + body.free() + return out + +fn main(): + port := 17521 + listener := tcp_listen("127.0.0.1", port) catch 0 + task := spawn serve_once(listener) + fd := tcp_connect("127.0.0.1", port) catch 0 + send_request(fd) + body := read_body(fd) + print(body.to_string_utf8() catch "?") + tcp_close(fd) + await task + tcp_close(listener) diff --git a/tests/expected/test_http2_empty_header_value.txt b/tests/expected/test_http2_empty_header_value.txt new file mode 100644 index 00000000..019713f7 --- /dev/null +++ b/tests/expected/test_http2_empty_header_value.txt @@ -0,0 +1 @@ +host: localhost|x-empty: |x-filled: value|x-trailing: | From 5688f87980920272d094bddbb1510ca4c6cc7773 Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 9 Aug 2026 14:34:56 +0000 Subject: [PATCH 3/4] perf: keep the redis line read small and the payload read large MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A socket read allocates and zeroes everything it asks for, so one read size for both jobs is the wrong trade: a 64 KiB read makes every "+PONG" pay for the one large bulk reply. Measured over 20,000 pings on one connection that cost about 5%. The line path is back to a 4 KiB read, matching what it always asked for, and the bulk gather — where the length is known and a read never over-reads — keeps the 64 KiB read that makes a large reply cheap. Small replies are level with the old client again and a 64 MiB bulk still lands in under 250 ms. --- std/redis.pith | 16 ++++++++++++---- 1 file changed, 12 insertions(+), 4 deletions(-) diff --git a/std/redis.pith b/std/redis.pith index f3cc3653..0c1af30a 100644 --- a/std/redis.pith +++ b/std/redis.pith @@ -22,8 +22,16 @@ MAX_BULK_LEN := 536870912 # 512 MiB — a single bulk string MAX_ARRAY_ITEMS := 10485760 # elements in one array reply MAX_REPLY_DEPTH := 64 # nested-array nesting -# bytes asked for in one socket read. -READ_CHUNK := 65536 +# bytes asked for in one socket read while looking for the end of a line. a +# read allocates and zeroes everything it asks for, so asking for more than a +# reply header needs makes every small reply pay for the one large one. +READ_CHUNK := 4096 + +# bytes asked for in one socket read while gathering a bulk payload of a known +# length. the size is already known here, so a larger read is a straight saving +# in syscalls and never over-reads: what is asked for is what is still missing, +# up to this. +PAYLOAD_CHUNK := 65536 # the most a bulk read reserves up front. the payload length comes off the # wire, so reserving all of it would let a twelve-byte header ask for half a @@ -163,8 +171,8 @@ impl Client: self.pos = 0 while got < need: mut want := need - got - if want > READ_CHUNK: - want = READ_CHUNK + if want > PAYLOAD_CHUNK: + want = PAYLOAD_CHUNK chunk := tcp_read_bytes(self.fd, want)! if chunk.len() == 0: fail "redis: connection closed" From 70078e726f8cd494192897665bdd4b09c9a60e4e Mon Sep 17 00:00:00 2001 From: Kacy Fortner Date: Sun, 9 Aug 2026 14:35:54 +0000 Subject: [PATCH 4/4] fix: say so when a redis buffered range is refused --- std/redis.pith | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/std/redis.pith b/std/redis.pith index 0c1af30a..f517cc30 100644 --- a/std/redis.pith +++ b/std/redis.pith @@ -166,7 +166,11 @@ impl Client: errdefer out.free() mut got := self.buffered() if got > 0: - out.write_range(self.buf, self.pos, got) + # the range is the unconsumed tail by construction, so a refusal + # here would mean the cursor and the buffer had come apart. say so + # rather than carry on having silently dropped the head of a reply. + if not out.write_range(self.buf, self.pos, got): + fail "redis: buffered range is outside the read buffer" self.buf = bytes.empty() self.pos = 0 while got < need: