diff --git a/bench/snippets/fetch-chunked-small.mjs b/bench/snippets/fetch-chunked-small.mjs new file mode 100644 index 000000000000..304d77cfa4ad --- /dev/null +++ b/bench/snippets/fetch-chunked-small.mjs @@ -0,0 +1,76 @@ +// Benchmark fetch() decoding small Transfer-Encoding: chunked bodies that +// arrive with their headers in a single read. Exercises the <=16 KiB +// dispatch in the HTTP client's chunked-body handler for both identity and +// gzip Content-Encoding. +import { bench, group, run } from "../runner.mjs"; +import net from "node:net"; +import zlib from "node:zlib"; + +function chunked(buf) { + return Buffer.concat([ + Buffer.from(buf.length.toString(16) + "\r\n"), + buf, + Buffer.from("\r\n0\r\n\r\n"), + ]); +} + +function makeReply(body, gzip) { + const payload = gzip ? zlib.gzipSync(body, { level: 6 }) : body; + return Buffer.concat([ + Buffer.from( + "HTTP/1.1 200 OK\r\n" + + "Transfer-Encoding: chunked\r\n" + + (gzip ? "Content-Encoding: gzip\r\n" : "") + + "Connection: keep-alive\r\n" + + "\r\n", + ), + chunked(payload), + ]); +} + +const sizes = [256, 4096, 15 * 1024]; +const bodies = Object.fromEntries(sizes.map(n => [n, Buffer.alloc(n, "x")])); +const replies = {}; +for (const n of sizes) { + replies[`i${n}`] = makeReply(bodies[n], false); + replies[`g${n}`] = makeReply(bodies[n], true); +} + +const server = net.createServer(sock => { + sock.setNoDelay(true); + let pending = Buffer.alloc(0); + sock.on("data", chunk => { + pending = Buffer.concat([pending, chunk]); + while (true) { + const end = pending.indexOf("\r\n\r\n"); + if (end < 0) break; + const head = pending.subarray(0, end).toString("latin1"); + pending = pending.subarray(end + 4); + const m = head.match(/^GET \/(\w+)/); + sock.write(replies[m[1]]); + } + }); +}); + +await new Promise(r => server.listen(0, r)); +const base = `http://127.0.0.1:${server.address().port}`; + +// Warm the keep-alive connection so the first iteration doesn't pay connect. +for (const key of Object.keys(replies)) await fetch(`${base}/${key}`).then(r => r.arrayBuffer()); + +for (const n of sizes) { + group(`chunked ${n}B`, () => { + bench("identity → arrayBuffer()", async () => { + const r = await fetch(`${base}/i${n}`); + await r.arrayBuffer(); + }); + bench("gzip → arrayBuffer()", async () => { + const r = await fetch(`${base}/g${n}`); + await r.arrayBuffer(); + }); + }); +} + +await run(); +server.close(); +process.exit(0); diff --git a/src/http/lib.rs b/src/http/lib.rs index 7288f064ab96..cd4d15513e7c 100644 --- a/src/http/lib.rs +++ b/src/http/lib.rs @@ -4781,8 +4781,17 @@ impl<'a> HTTPClient<'a> { &mut self, incoming_data: &[u8], ) -> crate::Result { + // The single-packet path decodes into a 16 KiB scratch so a small + // compressed body can be handed straight to `decompress_bytes` without + // ever allocating `compressed_body`. For identity bodies that shortcut + // doesn't exist: scratch copy + `append_slice_exact` is one memcpy more + // than `_from_multiple_packets`' append + decode-in-tail + truncate, so + // route identity bodies there unconditionally. let small_len = 16 * 1024usize; - if incoming_data.len() <= small_len && self.state.get_body_buffer().list.is_empty() { + if self.state.encoding.is_compressed() + && incoming_data.len() <= small_len + && self.state.get_body_buffer().list.is_empty() + { self.handle_response_body_chunked_encoding_from_single_packet(incoming_data) } else { self.handle_response_body_chunked_encoding_from_multiple_packets(incoming_data) diff --git a/test/js/web/fetch/fetch-gzip.test.ts b/test/js/web/fetch/fetch-gzip.test.ts index 701fbf4208d8..38ffa04837fa 100644 --- a/test/js/web/fetch/fetch-gzip.test.ts +++ b/test/js/web/fetch/fetch-gzip.test.ts @@ -1,6 +1,7 @@ import { Socket } from "bun"; import { beforeAll, describe, expect, it } from "bun:test"; import { bunEnv, bunExe, gcTick } from "harness"; +import { randomBytes } from "node:crypto"; import { once } from "node:events"; import { createServer } from "node:http"; import { createServer as createNetServer } from "node:net"; @@ -906,3 +907,51 @@ describe("empty compressed responses", () => { }); } }); + +// The <=16 KiB chunked-body dispatcher routes identity bodies through the +// append + decode-in-tail path and compressed bodies through the scratch +// decode path. These pin both paths for a body that fits in one read. +describe("small chunked body decode", () => { + function chunked(buf: Buffer): Buffer { + return Buffer.concat([Buffer.from(buf.length.toString(16) + "\r\n"), buf, Buffer.from("\r\n0\r\n\r\n")]); + } + + async function serve(extraHeader: string, body: Buffer, expected: Buffer): Promise { + const reply = Buffer.concat([ + Buffer.from("HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\nConnection: keep-alive\r\n" + extraHeader + "\r\n"), + chunked(body), + ]); + const sockets = new Set(); + const server = createNetServer(sock => { + sockets.add(sock); + sock.on("close", () => sockets.delete(sock)); + sock.on("error", () => {}); + sock.once("data", () => sock.write(reply)); + }); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const { port } = server.address() as import("node:net").AddressInfo; + try { + const res = await fetch(`http://127.0.0.1:${port}/`); + expect(res.status).toBe(200); + expect(Buffer.from(await res.arrayBuffer()).equals(expected)).toBe(true); + } finally { + for (const s of sockets) s.destroy(); + await new Promise(r => server.close(() => r())); + } + } + + for (const n of [1, 4096, 15 * 1024]) { + const body = Buffer.alloc(n, "x"); + it.concurrent(`identity ${n}B`, () => serve("", body, body)); + } + + // One compressed case as the unchanged-path control; randomBytes keeps the + // compressed size near the input size so the test exercises the 16 KiB gate + // with a multi-KiB payload rather than a ~50 B gzipped run of 'x'. + const raw = randomBytes(4096); + const gz = gzipSync(raw, { level: 6 }); + it.concurrent(`gzip ${gz.length}B compressed`, () => serve("Content-Encoding: gzip\r\n", gz, raw)); +});