From 06297d952d61af180e214f7d6923e6820202740a Mon Sep 17 00:00:00 2001 From: Matt Mastracci Date: Sat, 23 Sep 2023 08:55:28 -0600 Subject: feat(ext/web): use readableStreamDefaultReaderRead in resourceForReadableStream (#20622) We can go one level down in abstraction and avoid using the public `ReadableStream` APIs. This patch ~5% perf boost on small ReadableStream: ``` Running 10s test @ http://localhost:8080/ 2 threads and 10 connections Thread Stats Avg Stdev Max +/- Stdev Latency 148.32us 108.95us 3.88ms 95.71% Req/Sec 33.24k 2.68k 37.94k 73.76% 668188 requests in 10.10s, 77.74MB read Requests/sec: 66162.91 Transfer/sec: 7.70MB ``` main: ``` Running 10s test @ http://localhost:8080/ 2 threads and 10 connections Thread Stats Avg Stdev Max +/- Stdev Latency 150.23us 67.61us 4.39ms 94.80% Req/Sec 31.81k 1.55k 35.56k 83.17% 639078 requests in 10.10s, 74.36MB read Requests/sec: 63273.72 Transfer/sec: 7.36MB ``` --- cli/tests/unit/streams_test.ts | 58 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 58 insertions(+) (limited to 'cli/tests') diff --git a/cli/tests/unit/streams_test.ts b/cli/tests/unit/streams_test.ts index 3460e8ecd..e496b5c4f 100644 --- a/cli/tests/unit/streams_test.ts +++ b/cli/tests/unit/streams_test.ts @@ -60,6 +60,28 @@ function longStream() { }).pipeThrough(new TextEncoderStream()); } +// Long stream with Lorem Ipsum text. +// deno-lint-ignore no-explicit-any +function longAsyncStream(completion?: Deferred) { + let currentTimeout: number | undefined = undefined; + return new ReadableStream({ + async start(controller) { + for (let i = 0; i < 100; i++) { + await new Promise((r) => currentTimeout = setTimeout(r, 1)); + currentTimeout = undefined; + controller.enqueue(LOREM); + } + controller.close(); + }, + cancel(reason) { + completion?.resolve(reason); + if (currentTimeout !== undefined) { + clearTimeout(currentTimeout); + } + }, + }).pipeThrough(new TextEncoderStream()); +} + // Empty stream, closes either immediately or on a call to pull. function emptyStream(onPull: boolean) { return new ReadableStream({ @@ -104,6 +126,20 @@ function emptyChunkStream() { }); } +// Try to blow up any recursive reads. Note that because of the use of Array.shift in +// ReadableStream, this might not actually be able to complete with larger values of +// length. +function veryLongTinyPacketStream(length: number) { + return new ReadableStream({ + start(controller) { + for (let i = 0; i < length; i++) { + controller.enqueue(new Uint8Array([1])); + } + controller.close(); + }, + }); +} + // Creates a stream with the given number of packets, a configurable delay between packets, and a final // action (either "Throw" or "Close"). function makeStreamWithCount( @@ -179,6 +215,14 @@ Deno.test(async function readableStreamCloseWithoutRead() { assertEquals(await cancel, "resource closed"); }); +// Close the stream without reading anything +Deno.test(async function readableStreamCloseWithoutRead2() { + const cancel = deferred(); + const rid = resourceForReadableStream(longAsyncStream(cancel)); + core.ops.op_close(rid); + assertEquals(await cancel, "resource closed"); +}); + Deno.test(async function readableStreamPartial() { const rid = resourceForReadableStream(helloWorldStream()); const buffer = new Uint8Array(5); @@ -197,6 +241,20 @@ Deno.test(async function readableStreamLongReadAll() { core.ops.op_close(rid); }); +Deno.test(async function readableStreamLongAsyncReadAll() { + const rid = resourceForReadableStream(longAsyncStream()); + const buffer = await core.ops.op_read_all(rid); + assertEquals(buffer.length, LOREM.length * 100); + core.ops.op_close(rid); +}); + +Deno.test(async function readableStreamVeryLongReadAll() { + const rid = resourceForReadableStream(veryLongTinyPacketStream(10000)); + const buffer = await core.ops.op_read_all(rid); + assertEquals(buffer.length, 10000); + core.ops.op_close(rid); +}); + Deno.test(async function readableStreamLongByPiece() { const rid = resourceForReadableStream(longStream()); let total = 0; -- cgit v1.2.3