Skip to content

Commit a93077c

Browse files
committed
zlib: finish async convenience calls in one write
The asynchronous convenience methods (`zlib.deflate()`, `zlib.inflate()`, `zlib.gunzip()`, ...) wrote the input to a Transform stream and then ended it. That took two threadpool round trips, one for the data and one for the final flush, and ran the result through the stream machinery. The whole input is known up front, so unless `info` or a custom `flush` is requested, process it with the finish flush directly, the way the synchronous methods already do, and collect the output in the write callback. A call whose output fits in one buffer now takes a single threadpool round trip. The output is unchanged: the previous two-write sequence already produced the same bytes as the synchronous methods. `zstdCompress()` keeps the separate end-of-frame write, because a single `ZSTD_e_end` call can produce slightly larger frames. benchmark/compare.js, 40 runs, n=100000: confidence improvement zlib/inflate.js method='inflate' *** +94.91 % zlib/deflate.js method='deflate' *** +66.27 % Signed-off-by: Yagiz Nizipli <yagiz@nizipli.com> Assisted-by: Grok
1 parent 2052d0d commit a93077c

3 files changed

Lines changed: 454 additions & 24 deletions

File tree

‎lib/zlib.js‎

Lines changed: 71 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -87,6 +87,7 @@ const {
8787

8888
const kFlushFlag = Symbol('kFlushFlag');
8989
const kError = Symbol('kError');
90+
const kBufferResult = Symbol('kBufferResult');
9091

9192
const constants = internalBinding('constants').zlib;
9293
const {
@@ -124,7 +125,7 @@ for (const ckey of ObjectKeys(codes)) {
124125
codes[codes[ckey]] = ckey;
125126
}
126127

127-
function zlibBuffer(engine, buffer, callback) {
128+
function zlibBuffer(engine, buffer, callback, finishInOneWrite) {
128129
validateFunction(callback, 'callback');
129130
// Streams do not support non-Uint8Array ArrayBufferViews yet. Convert it to a
130131
// Buffer without copying.
@@ -136,44 +137,81 @@ function zlibBuffer(engine, buffer, callback) {
136137
engine.buffers = null;
137138
engine.nread = 0;
138139
engine.cb = callback;
139-
engine.on('data', zlibBufferOnData);
140140
engine.on('error', zlibBufferOnError);
141+
142+
// The whole input is known up front. Unless the engine is handed to the
143+
// caller (`info`) or a custom flush is requested, process it with the finish
144+
// flush directly, like the synchronous methods do. This takes one threadpool
145+
// round trip instead of a write followed by a separate final flush, and skips
146+
// the stream machinery.
147+
if (finishInOneWrite && !engine._info &&
148+
engine._defaultFlushFlag === FLUSH_BOUND[engine._flushBoundIdx][0]) {
149+
if (typeof buffer === 'string') {
150+
// Encode strings like Writable#write() does.
151+
buffer = Buffer.from(buffer, engine._writableState.defaultEncoding);
152+
}
153+
if (isUint8Array(buffer)) {
154+
engine[kBufferResult] = true;
155+
processChunk(engine, buffer, engine._finishFlushFlag,
156+
zlibBufferOnWriteDone);
157+
return;
158+
}
159+
}
160+
161+
engine.on('data', zlibBufferOnData);
141162
engine.on('end', zlibBufferOnEnd);
142163
engine.end(buffer);
143164
}
144165

145-
function zlibBufferOnData(chunk) {
146-
if (!this.buffers) {
147-
this.buffers = [chunk];
166+
function zlibBufferAddChunk(engine, chunk) {
167+
if (!engine.buffers) {
168+
engine.buffers = [chunk];
148169
} else {
149-
this.buffers.push(chunk);
170+
engine.buffers.push(chunk);
150171
}
151-
this.nread += chunk.length;
152-
if (this.nread > this._maxOutputLength) {
153-
this.close();
154-
this.removeAllListeners('end');
155-
this.cb(new ERR_BUFFER_TOO_LARGE(this._maxOutputLength));
172+
engine.nread += chunk.length;
173+
if (engine.nread > engine._maxOutputLength) {
174+
engine.close();
175+
engine.removeAllListeners('end');
176+
engine.cb(new ERR_BUFFER_TOO_LARGE(engine._maxOutputLength));
156177
}
157178
}
158179

180+
function zlibBufferOnData(chunk) {
181+
zlibBufferAddChunk(this, chunk);
182+
}
183+
159184
function zlibBufferOnError(err) {
160185
this.removeAllListeners('end');
161186
this.cb(err);
162187
}
163188

164-
function zlibBufferOnEnd() {
189+
function zlibBufferFinish(engine) {
165190
let buf;
166-
if (this.nread === 0) {
191+
if (engine.nread === 0) {
167192
buf = new FastBuffer();
168193
} else {
169-
const bufs = this.buffers;
170-
buf = (bufs.length === 1 ? bufs[0] : Buffer.concat(bufs, this.nread));
194+
const bufs = engine.buffers;
195+
buf = (bufs.length === 1 ? bufs[0] : Buffer.concat(bufs, engine.nread));
171196
}
172-
this.close();
173-
if (this._info)
174-
this.cb(null, { buffer: buf, engine: this });
197+
engine.close();
198+
if (engine._info)
199+
engine.cb(null, { buffer: buf, engine });
175200
else
176-
this.cb(null, buf);
201+
engine.cb(null, buf);
202+
}
203+
204+
function zlibBufferOnEnd() {
205+
zlibBufferFinish(this);
206+
}
207+
208+
// Called by processCallback(), with the handle as `this`, once the whole input
209+
// has been processed. Errors have already been reported to the callback, either
210+
// through the 'error' event or by zlibBufferAddChunk().
211+
function zlibBufferOnWriteDone(err) {
212+
const engine = this[owner_symbol];
213+
if (err === undefined && !engine.destroyed)
214+
zlibBufferFinish(engine);
177215
}
178216

179217
function zlibBufferSync(engine, buffer) {
@@ -552,7 +590,10 @@ function processCallback() {
552590
if (have > 0) {
553591
const out = self._outBuffer.slice(self._outOffset, self._outOffset + have);
554592
self._outOffset += have;
555-
streamBufferIsFull = !self.push(out);
593+
if (self[kBufferResult] === true)
594+
zlibBufferAddChunk(self, out);
595+
else
596+
streamBufferIsFull = !self.push(out);
556597
} else {
557598
assert(have === 0, 'have should not go down');
558599
}
@@ -618,7 +659,8 @@ function processCallback() {
618659
return;
619660
}
620661

621-
self.push(null);
662+
if (self[kBufferResult] !== true)
663+
self.push(null);
622664
}
623665

624666
// Finished with the chunk.
@@ -791,7 +833,8 @@ class Unzip extends Zlib {
791833
}
792834
}
793835

794-
function createConvenienceMethod(ctor, sync, prepareOpts) {
836+
function createConvenienceMethod(ctor, sync, prepareOpts,
837+
finishInOneWrite = true) {
795838
if (sync) {
796839
return function syncBufferWrapper(buffer, opts) {
797840
return zlibBufferSync(new ctor(opts), buffer);
@@ -805,7 +848,7 @@ function createConvenienceMethod(ctor, sync, prepareOpts) {
805848
if (prepareOpts !== undefined) {
806849
opts = prepareOpts(buffer, opts);
807850
}
808-
return zlibBuffer(new ctor(opts), buffer, callback);
851+
return zlibBuffer(new ctor(opts), buffer, callback, finishInOneWrite);
809852
};
810853
}
811854

@@ -1118,7 +1161,11 @@ module.exports = {
11181161
brotliCompressSync: createConvenienceMethod(BrotliCompress, true),
11191162
brotliDecompress: createConvenienceMethod(BrotliDecompress, false),
11201163
brotliDecompressSync: createConvenienceMethod(BrotliDecompress, true),
1121-
zstdCompress: createConvenienceMethod(ZstdCompress, false, withPledgedSrcSize),
1164+
// A zstd frame written by a single ZSTD_e_end call can be a few bytes larger
1165+
// than one written by ZSTD_e_continue followed by ZSTD_e_end, so
1166+
// zstdCompress() keeps writing the input and ending the frame separately.
1167+
zstdCompress: createConvenienceMethod(ZstdCompress, false, withPledgedSrcSize,
1168+
false),
11221169
zstdCompressSync: createConvenienceMethod(ZstdCompress, true),
11231170
zstdDecompress: createConvenienceMethod(ZstdDecompress, false),
11241171
zstdDecompressSync: createConvenienceMethod(ZstdDecompress, true),
Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
'use strict';
2+
// The asynchronous convenience methods know the whole input up front, so they
3+
// process it with the finish flush in a single write. Check that a call whose
4+
// output fits in one output buffer takes exactly one threadpool round trip.
5+
6+
const common = require('../common');
7+
const assert = require('assert');
8+
const async_hooks = require('async_hooks');
9+
const zlib = require('zlib');
10+
11+
const input = Buffer.from('hello world '.repeat(64));
12+
// [method, argument, function that turns the result back into `input`]
13+
const cases = [
14+
['deflate', input, zlib.inflateSync],
15+
['inflate', zlib.deflateSync(input)],
16+
['gzip', input, zlib.gunzipSync],
17+
['gunzip', zlib.gzipSync(input)],
18+
['unzip', zlib.gzipSync(input)],
19+
['deflateRaw', input, zlib.inflateRawSync],
20+
['inflateRaw', zlib.deflateRawSync(input)],
21+
['brotliCompress', input, zlib.brotliDecompressSync],
22+
['brotliDecompress', zlib.brotliCompressSync(input)],
23+
['zstdDecompress', zlib.zstdCompressSync(input)],
24+
];
25+
26+
// Counts the callbacks into JavaScript (`before` hooks) of each ZLIB resource
27+
// created while `fn` runs.
28+
async function countRoundTrips(fn) {
29+
const callbacks = new Map();
30+
const hook = async_hooks.createHook({
31+
init(asyncId, type) {
32+
if (type === 'ZLIB') callbacks.set(asyncId, 0);
33+
},
34+
before(asyncId) {
35+
if (callbacks.has(asyncId))
36+
callbacks.set(asyncId, callbacks.get(asyncId) + 1);
37+
},
38+
}).enable();
39+
try {
40+
await fn();
41+
} finally {
42+
hook.disable();
43+
}
44+
return [...callbacks.values()];
45+
}
46+
47+
function callAsync(method, buffer, options = {}) {
48+
return new Promise((resolve, reject) => {
49+
zlib[method](buffer, options, common.mustCall((err, result) => {
50+
if (err) reject(err);
51+
else resolve(result);
52+
}));
53+
});
54+
}
55+
56+
(async () => {
57+
for (const [method, buffer, undo = (result) => result] of cases) {
58+
let result;
59+
const roundTrips = await countRoundTrips(async () => {
60+
result = await callAsync(method, buffer);
61+
});
62+
assert.deepStrictEqual(roundTrips, [1], method);
63+
assert.deepStrictEqual(undo(result), input, method);
64+
}
65+
66+
// Errors are also reported after a single round trip.
67+
let error;
68+
const roundTrips = await countRoundTrips(async () => {
69+
error = await callAsync('inflate', Buffer.from('not zlib data'))
70+
.catch((err) => err);
71+
});
72+
assert.deepStrictEqual(roundTrips, [1]);
73+
assert.strictEqual(error.code, 'Z_DATA_ERROR');
74+
})().then(common.mustCall());

0 commit comments

Comments
 (0)