diff --git a/benchmark/webstreams/compression.js b/benchmark/webstreams/compression.js new file mode 100644 index 000000000000..8bc0fdc6225c --- /dev/null +++ b/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); +} diff --git a/benchmark/webstreams/encoding.js b/benchmark/webstreams/encoding.js new file mode 100644 index 000000000000..b548e5a69b05 --- /dev/null +++ b/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); +} diff --git a/lib/internal/webstreams/adapters.js b/lib/internal/webstreams/adapters.js index 42e1221bb307..8488bd04424f 100644 --- a/lib/internal/webstreams/adapters.js +++ b/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, }; diff --git a/lib/internal/webstreams/compression.js b/lib/internal/webstreams/compression.js index 70359d1158ee..ee00a64a501a 100644 --- a/lib/internal/webstreams/compression.js +++ b/lib/internal/webstreams/compression.js @@ -1,52 +1,163 @@ 'use strict'; const { + ArrayPrototypeFilter, + MathMin, ObjectDefineProperties, + ObjectKeys, + Promise, + SafeSet, + StringPrototypeStartsWith, SymbolToStringTag, + TypeError, + TypedArrayPrototypeGetByteLength, + TypedArrayPrototypeSet, + TypedArrayPrototypeSubarray, + Uint32Array, + Uint8Array, } = primordials; const { - newReadableWritablePairFromDuplex, - kValidateChunk, - kDestroyOnSyncError, -} = require('internal/webstreams/adapters'); + TransformStream, +} = require('internal/webstreams/transformstream'); const { customInspect } = require('internal/webstreams/util'); const { isArrayBufferView, + isAnyArrayBuffer, isSharedArrayBuffer, + isUint8Array, } = require('internal/util/types'); const { customInspectSymbol: kInspect, kEnumerableProperty, + setOwnProperty, } = require('internal/util'); const { codes: { ERR_INVALID_ARG_TYPE, + ERR_STREAM_NULL_VALUES, + ERR_TRAILING_JUNK_AFTER_STREAM_END, }, + genericNodeError, } = require('internal/errors'); const { createEnumConverter } = require('internal/webidl'); -let zlib; -function lazyZlib() { - zlib ??= require('zlib'); - return zlib; +const { + Zlib, + BrotliDecoder, + BrotliEncoder, +} = internalBinding('zlib'); + +const { zlib: constants } = internalBinding('constants'); +const { + BROTLI_DECODE, + BROTLI_ENCODE, + BROTLI_OPERATION_FINISH, + BROTLI_OPERATION_PROCESS, + DEFLATE, + DEFLATERAW, + GUNZIP, + GZIP, + INFLATE, + INFLATERAW, + Z_DEFAULT_COMPRESSION, + Z_DEFAULT_MEMLEVEL, + Z_DEFAULT_STRATEGY, + Z_DEFAULT_WINDOWBITS, + Z_FINISH, + Z_NO_FLUSH, +} = constants; + +const { Buffer } = require('buffer'); + +let setImmediate; + +// Output is accumulated in fixed-size buffers and emitted as soon as it is +// produced, mirroring how the zlib streams emit their output. Produced +// regions at least half a buffer large are emitted as zero-copy views (and +// the buffer is retired so no two emitted chunks ever share memory); +// smaller regions are copied out so that tiny chunks do not retain large +// allocations. +const kOutputBufferSize = 65536; +const kEmitViewThreshold = kOutputBufferSize / 2; + +// Inputs larger than this are processed in slices, with a turn of the event +// loop in between, so that compressing/decompressing a huge chunk does not +// block the event loop for its full duration. +const kInputSliceSize = 65536; + +// Collect all negative (error) ZLIB codes and Z_NEED_DICT. +const ZLIB_FAILURES = new SafeSet( + ArrayPrototypeFilter( + ObjectKeys(constants), + (code) => code === 'Z_NEED_DICT' || constants[code] < 0, + ), +); + +// Compression error codes are surfaced as TypeError to align with the +// WHATWG Compression Streams specification. +function convertToTypeError(message, errno, code) { + const cause = genericNodeError(message, { errno, code }); + cause.errno = errno; + cause.code = code; + if (ZLIB_FAILURES.has(code) || + // Brotli decoder error codes are formatted as 'ERR_' + + // BrotliDecoderErrorString(), where the latter returns strings like + // '_ERROR_FORMAT_...', '_ERROR_ALLOC_...', '_ERROR_UNREACHABLE', etc. + // The resulting JS error codes all start with 'ERR__ERROR_'. + StringPrototypeStartsWith(code, 'ERR__ERROR_')) { + // eslint-disable-next-line no-restricted-syntax + const error = new TypeError(undefined, { cause }); + setOwnProperty(error, 'code', code); + return error; + } + return cause; } // Per the Compression Streams spec, chunks must be BufferSource // (ArrayBuffer or ArrayBufferView not backed by SharedArrayBuffer). -function validateBufferSourceChunk(chunk) { - if (isSharedArrayBuffer(isArrayBufferView(chunk) ? chunk.buffer : chunk)) { - throw new ERR_INVALID_ARG_TYPE( - 'chunk', - ['ArrayBuffer', 'Buffer', 'TypedArray', 'DataView'], - chunk, - ); +// Additionally, strings are accepted for backwards compatibility with the +// previous Node.js streams-based implementation. +function normalizeChunk(chunk) { + if (chunk === null) { + throw new ERR_STREAM_NULL_VALUES(); + } + if (typeof chunk === 'string') { + return Buffer.from(chunk); + } + if (isArrayBufferView(chunk)) { + if (isSharedArrayBuffer(chunk.buffer)) { + throw new ERR_INVALID_ARG_TYPE( + 'chunk', + ['ArrayBuffer', 'Buffer', 'TypedArray', 'DataView'], + chunk, + ); + } + if (isUint8Array(chunk)) { + return chunk; + } + return new Uint8Array(chunk.buffer, chunk.byteOffset, chunk.byteLength); + } + if (isAnyArrayBuffer(chunk)) { + if (isSharedArrayBuffer(chunk)) { + throw new ERR_INVALID_ARG_TYPE( + 'chunk', + ['ArrayBuffer', 'Buffer', 'TypedArray', 'DataView'], + chunk, + ); + } + return new Uint8Array(chunk); } + throw new ERR_INVALID_ARG_TYPE( + 'chunk', + ['string', 'Buffer', 'TypedArray', 'DataView'], + chunk, + ); } const formatConverter = createEnumConverter('CompressionFormat', [ @@ -56,13 +167,242 @@ const formatConverter = createEnumConverter('CompressionFormat', [ 'brotli', ]); +// Processes chunks synchronously on the current thread using the raw zlib +// (or brotli) handle, avoiding both the threadpool round trip that the +// zlib streams make for every write and the stream.Duplex adapter layers. +// For the small chunks that flow through streams pipelines, running the +// compression inline is significantly cheaper than dispatching each chunk +// to the threadpool and waiting for the event loop to observe completion. +class CompressionHandle { + #handle; + #writeState = new Uint32Array(2); + #error; + #closed = false; + #outBuffer = null; + #outOffset = 0; + #noFlushFlag; + #finishFlag; + #rejectTrailingInput; + + constructor(mode) { + const onerror = (message, errno, code) => { + this.#error = convertToTypeError(message, errno, code); + }; + if (mode === BROTLI_ENCODE || mode === BROTLI_DECODE) { + this.#handle = mode === BROTLI_DECODE ? + new BrotliDecoder(mode) : new BrotliEncoder(mode); + this.#handle.init( + getBrotliDefaultParams(), + this.#writeState, + noopOnWriteComplete, + undefined, + ); + this.#noFlushFlag = BROTLI_OPERATION_PROCESS; + this.#finishFlag = BROTLI_OPERATION_FINISH; + this.#rejectTrailingInput = mode === BROTLI_DECODE; + } else { + const decompress = + mode === INFLATE || mode === INFLATERAW || mode === GUNZIP; + // A windowBits value of 0 tells zlib to use the window size stored + // in the header of the compressed stream. + const windowBits = mode === INFLATE || mode === GUNZIP ? + 0 : Z_DEFAULT_WINDOWBITS; + this.#handle = new Zlib(mode); + this.#handle.init( + windowBits, + Z_DEFAULT_COMPRESSION, + Z_DEFAULT_MEMLEVEL, + Z_DEFAULT_STRATEGY, + this.#writeState, + noopOnWriteComplete, + undefined, + decompress, + ); + this.#noFlushFlag = Z_NO_FLUSH; + this.#finishFlag = Z_FINISH; + this.#rejectTrailingInput = decompress; + } + this.#handle.onerror = onerror; + } + + transform(chunk, controller) { + chunk = normalizeChunk(chunk); + if (TypedArrayPrototypeGetByteLength(chunk) <= kInputSliceSize) { + this.#process(chunk, this.#noFlushFlag, controller); + return; + } + return this.#processSlices(chunk, controller); + } + + async #processSlices(chunk, controller) { + let offset = 0; + while (offset < TypedArrayPrototypeGetByteLength(chunk)) { + if (this.#closed) return; + const end = MathMin( + offset + kInputSliceSize, TypedArrayPrototypeGetByteLength(chunk)); + this.#process( + TypedArrayPrototypeSubarray(chunk, offset, end), + this.#noFlushFlag, + controller); + offset = end; + if (offset < TypedArrayPrototypeGetByteLength(chunk)) { + setImmediate ??= require('timers').setImmediate; + await new Promise(setImmediate); + } + } + } + + flush(controller) { + try { + this.#process(kEmptyInput, this.#finishFlag, controller); + } finally { + this.close(); + } + } + + close() { + if (!this.#closed) { + this.#closed = true; + this.#handle.close(); + } + } + + #process(chunk, flushFlag, controller) { + let availIn = TypedArrayPrototypeGetByteLength(chunk); + let inOff = 0; + const writeState = this.#writeState; + const handle = this.#handle; + + let availOutAfter; + let availInAfter; + do { + if (this.#outBuffer === null) { + this.#outBuffer = new Uint8Array(kOutputBufferSize); + this.#outOffset = 0; + } + const availOutBefore = kOutputBufferSize - this.#outOffset; + handle.writeSync(flushFlag, + chunk, // in + inOff, // in_off + availIn, // in_len + this.#outBuffer, // out + this.#outOffset, // out_off + availOutBefore); // out_len + if (this.#error !== undefined) { + const error = this.#error; + this.close(); + throw error; + } + + availOutAfter = writeState[0]; + availInAfter = writeState[1]; + + const have = availOutBefore - availOutAfter; + if (have > 0) { + this.#emit(have, controller); + } + + // Exhausted the output buffer: emit and reprocess the rest of the + // input against a fresh buffer. + inOff += availIn - availInAfter; + availIn = availInAfter; + } while (availOutAfter === 0); + + if (availInAfter > 0 && this.#rejectTrailingInput) { + // The compression library was not interested in receiving more data: + // the compressed stream has ended, with junk data trailing behind it. + const error = new ERR_TRAILING_JUNK_AFTER_STREAM_END(); + this.close(); + throw error; + } + } + + #emit(have, controller) { + const offset = this.#outOffset; + const buffer = this.#outBuffer; + let chunk; + if (have >= kEmitViewThreshold) { + // Emit a zero-copy view and retire the buffer so that no two emitted + // chunks ever share the same backing memory. + chunk = new Uint8Array(buffer.buffer, offset, have); + this.#outBuffer = null; + } else { + chunk = new Uint8Array(have); + TypedArrayPrototypeSet( + chunk, TypedArrayPrototypeSubarray(buffer, offset, offset + have)); + this.#outOffset = offset + have; + if (this.#outOffset === kOutputBufferSize) { + this.#outBuffer = null; + } + } + controller.enqueue(chunk); + } +} + +const kEmptyInput = new Uint8Array(0); + +// The write callback is required by the handle's init function, but it is +// only ever invoked by asynchronous writes, which are never issued here. +function noopOnWriteComplete() {} + +let brotliDefaultParams; +function getBrotliDefaultParams() { + if (brotliDefaultParams === undefined) { + let maxParam = 0; + for (const key of ObjectKeys(constants)) { + if (StringPrototypeStartsWith(key, 'BROTLI_PARAM_') && + constants[key] > maxParam) { + maxParam = constants[key]; + } + } + // -1 (as an unsigned 32-bit value) marks a parameter as unset. + brotliDefaultParams = new Uint32Array(maxParam + 1); + brotliDefaultParams.fill(-1); + } + return brotliDefaultParams; +} + +// These match the strategies that the previous stream.Duplex-based +// implementation derived from the zlib streams' high water marks. +function getWritableStrategy() { + return { + highWaterMark: 16384, + size(chunk) { + return chunk?.byteLength ?? chunk?.length ?? 1; + }, + }; +} + +function getReadableStrategy() { + return { + highWaterMark: 16384, + size(chunk) { + return chunk.byteLength; + }, + }; +} + +function createTransform(mode) { + const handle = new CompressionHandle(mode); + return new TransformStream({ + transform(chunk, controller) { + return handle.transform(chunk, controller); + }, + flush(controller) { + handle.flush(controller); + }, + cancel() { + handle.close(); + }, + }, getWritableStrategy(), getReadableStrategy()); +} + /** * @typedef {import('./readablestream').ReadableStream} ReadableStream * @typedef {import('./writablestream').WritableStream} WritableStream */ class CompressionStream { - #handle; #transform; /** @@ -75,22 +415,18 @@ class CompressionStream { }); switch (format) { case 'deflate': - this.#handle = lazyZlib().createDeflate(); + this.#transform = createTransform(DEFLATE); break; case 'deflate-raw': - this.#handle = lazyZlib().createDeflateRaw(); + this.#transform = createTransform(DEFLATERAW); break; case 'gzip': - this.#handle = lazyZlib().createGzip(); + this.#transform = createTransform(GZIP); break; case 'brotli': - this.#handle = lazyZlib().createBrotliCompress(); + this.#transform = createTransform(BROTLI_ENCODE); break; } - this.#transform = newReadableWritablePairFromDuplex(this.#handle, { - [kValidateChunk]: validateBufferSourceChunk, - [kDestroyOnSyncError]: true, - }); } /** @@ -118,7 +454,6 @@ class CompressionStream { } class DecompressionStream { - #handle; #transform; /** @@ -131,30 +466,18 @@ class DecompressionStream { }); switch (format) { case 'deflate': - this.#handle = lazyZlib().createInflate({ - rejectGarbageAfterEnd: true, - }); + this.#transform = createTransform(INFLATE); break; case 'deflate-raw': - this.#handle = lazyZlib().createInflateRaw({ - rejectGarbageAfterEnd: true, - }); + this.#transform = createTransform(INFLATERAW); break; case 'gzip': - this.#handle = lazyZlib().createGunzip({ - rejectGarbageAfterEnd: true, - }); + this.#transform = createTransform(GUNZIP); break; case 'brotli': - this.#handle = lazyZlib().createBrotliDecompress({ - rejectGarbageAfterEnd: true, - }); + this.#transform = createTransform(BROTLI_DECODE); break; } - this.#transform = newReadableWritablePairFromDuplex(this.#handle, { - [kValidateChunk]: validateBufferSourceChunk, - [kDestroyOnSyncError]: true, - }); } /** diff --git a/lib/internal/webstreams/encoding.js b/lib/internal/webstreams/encoding.js index f316222ccbf0..8e3af430d1a9 100644 --- a/lib/internal/webstreams/encoding.js +++ b/lib/internal/webstreams/encoding.js @@ -4,6 +4,7 @@ const { ObjectDefineProperties, String, StringPrototypeCharCodeAt, + StringPrototypeSlice, Uint8Array, } = primordials; @@ -46,32 +47,30 @@ class TextEncoderStream { this.#transform = new TransformStream({ transform: (chunk, controller) => { // https://encoding.spec.whatwg.org/#encode-and-enqueue-a-chunk + // + // The spec describes a per-code-unit loop whose only observable + // effects are (a) holding back a high surrogate that ends a chunk + // so it can be paired with a low surrogate starting the next chunk + // and (b) replacing unpaired surrogates with U+FFFD. The encoder + // already performs (b), so only the chunk boundaries need special + // handling here. chunk = String(chunk); - let finalChunk = ''; - for (let i = 0; i < chunk.length; i++) { - const item = chunk[i]; - const codeUnit = StringPrototypeCharCodeAt(item, 0); - if (this.#pendingHighSurrogate !== null) { - const highSurrogate = this.#pendingHighSurrogate; - this.#pendingHighSurrogate = null; - if (0xDC00 <= codeUnit && codeUnit <= 0xDFFF) { - finalChunk += highSurrogate + item; - continue; - } - finalChunk += '\uFFFD'; - } - if (0xD800 <= codeUnit && codeUnit <= 0xDBFF) { - this.#pendingHighSurrogate = item; - continue; - } - if (0xDC00 <= codeUnit && codeUnit <= 0xDFFF) { - finalChunk += '\uFFFD'; - continue; - } - finalChunk += item; + if (chunk.length === 0) { + return; } - if (finalChunk) { - const value = this.#handle.encode(finalChunk); + if (this.#pendingHighSurrogate !== null) { + chunk = this.#pendingHighSurrogate + chunk; + this.#pendingHighSurrogate = null; + } + const lastCodeUnit = StringPrototypeCharCodeAt(chunk, chunk.length - 1); + if (0xD800 <= lastCodeUnit && lastCodeUnit <= 0xDBFF) { + // A high surrogate at the end of the chunk may pair with a low + // surrogate at the start of the next one: hold it back. + this.#pendingHighSurrogate = StringPrototypeSlice(chunk, -1); + chunk = StringPrototypeSlice(chunk, 0, -1); + } + if (chunk) { + const value = this.#handle.encode(chunk); controller.enqueue(value); } }, diff --git a/test/parallel/test-webstreams-adapters-sync-write-error.js b/test/parallel/test-webstreams-adapters-sync-write-error.js index 748f682365ee..7123ed2e3e71 100644 --- a/test/parallel/test-webstreams-adapters-sync-write-error.js +++ b/test/parallel/test-webstreams-adapters-sync-write-error.js @@ -11,8 +11,7 @@ const { // Verify that when the underlying Node.js stream throws synchronously from // write(), the writable web stream properly rejects but does not destroy -// the stream (destroy-on-sync-throw is only used internally by -// CompressionStream/DecompressionStream). +// the stream. test('WritableStream from Node.js stream handles sync write throw', async () => { const error = new TypeError('invalid chunk');