stream: cut per-chunk overhead in web streams pipelines

benchmark/webstreams/compression.js

@@ -0,0 +1,67 @@
+'use strict';
+const common = require('../common.js');
+const { gzipSync, deflateSync } = require('node:zlib');
+const assert = require('node:assert');
+
+const bench = common.createBenchmark(main, {
+  chunkSize: [4096, 65536],
+  totalBytes: [64 << 20],
+  kind: ['compress', 'decompress'],
+  format: ['gzip', 'deflate'],
+});
+
+// Repetitive but non-trivial payload.
+function makePayload(totalBytes, chunkSize) {
+  const line = 'time=2024-01-01T00:00:00.000Z level=info request=42 ' +
+    'method=GET path=/api/v1/items status=200 duration=13ms\n';
+  const chunk = Buffer.alloc(chunkSize, line);
+  const chunks = [];
+  for (let n = 0; n < totalBytes; n += chunk.length)
+    chunks.push(chunk);
+  return chunks;
+}
+
+function makeSource(chunks) {
+  let i = 0;
+  return new ReadableStream({
+    pull(controller) {
+      if (i < chunks.length) {
+        controller.enqueue(chunks[i++]);
+      } else {
+        controller.close();
+      }
+    },
+  });
+}
+
+async function main({ chunkSize, totalBytes, kind, format }) {
+  let chunks = makePayload(totalBytes, chunkSize);
+
+  let stream;
+  if (kind === 'compress') {
+    stream = new CompressionStream(format);
+  } else {
+    // Rechunk the compressed payload so the decompressor sees the
+    // configured chunk size.
+    const compress = format === 'gzip' ? gzipSync : deflateSync;
+    const compressed = compress(Buffer.concat(chunks));
+    chunks = [];
+    for (let off = 0; off < compressed.length; off += chunkSize)
+      chunks.push(compressed.subarray(off, off + chunkSize));
+    stream = new DecompressionStream(format);
+  }
+
+  const source = makeSource(chunks);
+
+  let outputBytes = 0;
+  bench.start();
+  for await (const chunk of source.pipeThrough(stream)) {
+    outputBytes += chunk.byteLength;
+  }
+  bench.end(totalBytes / (1024 * 1024));
+
+  if (kind === 'decompress')
+    assert.strictEqual(outputBytes, totalBytes);
+  else
+    assert.notStrictEqual(outputBytes, 0);
+}

benchmark/webstreams/encoding.js

@@ -0,0 +1,52 @@
+'use strict';
+const common = require('../common.js');
+const assert = require('node:assert');
+
+const bench = common.createBenchmark(main, {
+  chunkSize: [4096],
+  totalBytes: [64 << 20],
+  kind: ['encode', 'decode', 'transcode'],
+});
+
+async function main({ chunkSize, totalBytes, kind }) {
+  const line = 'time=2024-01-01T00:00:00.000Z level=info request=42 ' +
+    'method=GET path=/api/v1/items status=200 duration=13ms\n';
+  const byteChunk = Buffer.alloc(chunkSize, line);
+  const stringChunk = byteChunk.toString();
+  const nChunks = Math.ceil(totalBytes / chunkSize);
+
+  let i = 0;
+  const input = kind === 'encode' ? stringChunk : byteChunk;
+  const source = new ReadableStream({
+    pull(controller) {
+      if (i++ < nChunks) {
+        controller.enqueue(input);
+      } else {
+        controller.close();
+      }
+    },
+  });
+
+  let stream = source;
+  switch (kind) {
+    case 'encode':
+      stream = source.pipeThrough(new TextEncoderStream());
+      break;
+    case 'decode':
+      stream = source.pipeThrough(new TextDecoderStream());
+      break;
+    case 'transcode':
+      stream = source
+        .pipeThrough(new TextDecoderStream())
+        .pipeThrough(new TextEncoderStream());
+      break;
+  }
+
+  let processed = 0;
+  bench.start();
+  for await (const chunk of stream) {
+    processed += chunk.length;
+  }
+  bench.end(totalBytes / (1024 * 1024));
+  assert.strictEqual(processed, nChunks * chunkSize);
+}

lib/internal/webstreams/adapters.js

@@ -100,9 +100,6 @@ const { UV_EOF } = internalBinding('uv');
 
 const encoder = new TextEncoder();
 
-const kValidateChunk = Symbol('kValidateChunk');
-const kDestroyOnSyncError = Symbol('kDestroyOnSyncError');
-
 // Collect all negative (error) ZLIB codes and Z_NEED_DICT
 const ZLIB_FAILURES = new SafeSet(
   ArrayPrototypeFilter(
@@ -236,32 +233,18 @@ function newWritableStreamFromStreamWritable(streamWritable, options = kEmptyObj
     start(c) { controller = c; },
 
     write(chunk) {
-      try {
-        options[kValidateChunk]?.(chunk);
-        if (!streamWritable.writableObjectMode && isAnyArrayBuffer(chunk)) {
-          chunk = new Uint8Array(chunk);
-        }
-        if (streamWritable.writableNeedDrain || !streamWritable.write(chunk)) {
-          backpressurePromise = PromiseWithResolvers();
-          if (!streamWritable.writableNeedDrain) {
-            backpressurePromise.resolve();
-          }
-          return SafePromisePrototypeFinally(
-            backpressurePromise.promise, () => {
-              backpressurePromise = undefined;
-            });
-        }
-      } catch (error) {
-        // When the kDestroyOnSyncError flag is set (e.g. for
-        // CompressionStream), a sync throw must also destroy the
-        // stream so the readable side is errored too. Without this
-        // the readable side hangs forever. This replicates the
-        // TransformStream semantics: error both sides on any throw
-        // in the transform path.
-        if (options[kDestroyOnSyncError]) {
-          destroy(streamWritable, error);
+      if (!streamWritable.writableObjectMode && isAnyArrayBuffer(chunk)) {
+        chunk = new Uint8Array(chunk);
+      }
+      if (streamWritable.writableNeedDrain || !streamWritable.write(chunk)) {
+        backpressurePromise = PromiseWithResolvers();
+        if (!streamWritable.writableNeedDrain) {
+          backpressurePromise.resolve();
         }
-        throw error;
+        return SafePromisePrototypeFinally(
+          backpressurePromise.promise, () => {
+            backpressurePromise = undefined;
+          });
       }
     },
 
@@ -694,15 +677,9 @@ function newReadableWritablePairFromDuplex(duplex, options = kEmptyObject) {
     return { readable, writable };
   }
 
-  const writableOptions = {
-    __proto__: null,
-    [kValidateChunk]: options[kValidateChunk],
-    [kDestroyOnSyncError]: options[kDestroyOnSyncError],
-  };
-
   const writable =
     isWritable(duplex) ?
-      newWritableStreamFromStreamWritable(duplex, writableOptions) :
+      newWritableStreamFromStreamWritable(duplex) :
       new WritableStream();
 
   if (!isWritable(duplex))
@@ -1093,6 +1070,4 @@ module.exports = {
   newStreamDuplexFromReadableWritablePair,
   newWritableStreamFromStreamBase,
   newReadableStreamFromStreamBase,
-  kValidateChunk,
-  kDestroyOnSyncError,
 };