From 87358944993e557fdd5af72545f8e848bf44c8a9 Mon Sep 17 00:00:00 2001 From: Luigi Pinca Date: Thu, 6 Jan 2022 17:01:04 +0100 Subject: [PATCH] [perf] Reduce buffer allocations Do not convert strings to `Buffer`s if data does not need to be masked. Refs: https://github.com/websockets/ws/pull/1998 --- lib/permessage-deflate.js | 4 +- lib/sender.js | 235 ++++++++++++++++++++------------------ test/sender.test.js | 29 +++-- 3 files changed, 149 insertions(+), 119 deletions(-) diff --git a/lib/permessage-deflate.js b/lib/permessage-deflate.js index 504069719..94603c98d 100644 --- a/lib/permessage-deflate.js +++ b/lib/permessage-deflate.js @@ -313,7 +313,7 @@ class PerMessageDeflate { /** * Compress data. Concurrency limited. * - * @param {Buffer} data Data to compress + * @param {(Buffer|String)} data Data to compress * @param {Boolean} fin Specifies whether or not this is the last fragment * @param {Function} callback Callback * @public @@ -395,7 +395,7 @@ class PerMessageDeflate { /** * Compress data. * - * @param {Buffer} data Data to compress + * @param {(Buffer|String)} data Data to compress * @param {Boolean} fin Specifies whether or not this is the last fragment * @param {Function} callback Callback * @private diff --git a/lib/sender.js b/lib/sender.js index 2417656d7..f5aec6980 100644 --- a/lib/sender.js +++ b/lib/sender.js @@ -46,8 +46,11 @@ class Sender { /** * Frames a piece of data according to the HyBi WebSocket protocol. * - * @param {Buffer} data The data to frame + * @param {(Buffer|String)} data The data to frame. The string type is only + * allowed for internal usage. * @param {Object} options Options object + * @param {Number} [options.byteLength] The length in bytes of the data. Only + * relevant if data is a string. This is only for internal usage. * @param {Boolean} [options.fin=false] Specifies whether or not to set the * FIN bit * @param {Function} [options.generateMask] The function used to generate the @@ -61,7 +64,7 @@ class Sender { * modified * @param {Boolean} [options.rsv1=false] Specifies whether or not to set the * RSV1 bit - * @return {Buffer[]} The framed data as a list of `Buffer` instances + * @return {(Buffer|String)[]} The framed data * @public */ static frame(data, options) { @@ -80,12 +83,27 @@ class Sender { } skipMasking = (mask[0] | mask[1] | mask[2] | mask[3]) === 0; - if (options.readOnly && !skipMasking) merge = true; - offset = 6; } - let payloadLength = data.length; + let dataLength; + + if (typeof data === 'string') { + if (!options.mask || skipMasking) { + dataLength = + options.byteLength !== undefined + ? options.byteLength + : Buffer.byteLength(data); + } else { + data = Buffer.from(data); + dataLength = data.length; + } + } else { + dataLength = data.length; + merge = options.mask && options.readOnly && !skipMasking; + } + + let payloadLength = dataLength; if (data.length >= 65536) { offset += 8; @@ -95,7 +113,7 @@ class Sender { payloadLength = 126; } - const target = Buffer.allocUnsafe(merge ? data.length + offset : offset); + const target = Buffer.allocUnsafe(merge ? dataLength + offset : offset); target[0] = options.fin ? options.opcode | 0x80 : options.opcode; if (options.rsv1) target[0] |= 0x40; @@ -103,10 +121,10 @@ class Sender { target[1] = payloadLength; if (payloadLength === 126) { - target.writeUInt16BE(data.length, 2); + target.writeUInt16BE(dataLength, 2); } else if (payloadLength === 127) { target[2] = target[3] = 0; - target.writeUIntBE(data.length, 4, 6); + target.writeUIntBE(dataLength, 4, 6); } if (!options.mask) return [target, data]; @@ -164,36 +182,24 @@ class Sender { } } + const options = { + byteLength: buf.length, + fin: true, + generateMask: this._generateMask, + mask, + maskBuffer: this._maskBuffer, + opcode: 0x08, + readOnly: false, + rsv1: false + }; + if (this._deflating) { - this.enqueue([this.doClose, buf, mask, cb]); + this.enqueue([this.dispatch, buf, false, options, cb]); } else { - this.doClose(buf, mask, cb); + this.sendFrame(Sender.frame(buf, options), cb); } } - /** - * Frames and sends a close message. - * - * @param {Buffer} data The message to send - * @param {Boolean} [mask=false] Specifies whether or not to mask `data` - * @param {Function} [cb] Callback - * @private - */ - doClose(data, mask, cb) { - this.sendFrame( - Sender.frame(data, { - fin: true, - rsv1: false, - opcode: 0x08, - mask, - maskBuffer: this._maskBuffer, - generateMask: this._generateMask, - readOnly: false - }), - cb - ); - } - /** * Sends a ping message to the other peer. * @@ -203,43 +209,40 @@ class Sender { * @public */ ping(data, mask, cb) { - const buf = toBuffer(data); + let byteLength; + let readOnly; + + if (typeof data === 'string') { + byteLength = Buffer.byteLength(data); + readOnly = false; + } else { + data = toBuffer(data); + byteLength = data.length; + readOnly = toBuffer.readOnly; + } - if (buf.length > 125) { + if (byteLength > 125) { throw new RangeError('The data size must not be greater than 125 bytes'); } + const options = { + byteLength, + fin: true, + generateMask: this._generateMask, + mask, + maskBuffer: this._maskBuffer, + opcode: 0x09, + readOnly, + rsv1: false + }; + if (this._deflating) { - this.enqueue([this.doPing, buf, mask, toBuffer.readOnly, cb]); + this.enqueue([this.dispatch, data, false, options, cb]); } else { - this.doPing(buf, mask, toBuffer.readOnly, cb); + this.sendFrame(Sender.frame(data, options), cb); } } - /** - * Frames and sends a ping message. - * - * @param {Buffer} data The message to send - * @param {Boolean} [mask=false] Specifies whether or not to mask `data` - * @param {Boolean} [readOnly=false] Specifies whether `data` can be modified - * @param {Function} [cb] Callback - * @private - */ - doPing(data, mask, readOnly, cb) { - this.sendFrame( - Sender.frame(data, { - fin: true, - rsv1: false, - opcode: 0x09, - mask, - maskBuffer: this._maskBuffer, - generateMask: this._generateMask, - readOnly - }), - cb - ); - } - /** * Sends a pong message to the other peer. * @@ -249,43 +252,40 @@ class Sender { * @public */ pong(data, mask, cb) { - const buf = toBuffer(data); + let byteLength; + let readOnly; + + if (typeof data === 'string') { + byteLength = Buffer.byteLength(data); + readOnly = false; + } else { + data = toBuffer(data); + byteLength = data.length; + readOnly = toBuffer.readOnly; + } - if (buf.length > 125) { + if (byteLength > 125) { throw new RangeError('The data size must not be greater than 125 bytes'); } + const options = { + byteLength, + fin: true, + generateMask: this._generateMask, + mask, + maskBuffer: this._maskBuffer, + opcode: 0x0a, + readOnly, + rsv1: false + }; + if (this._deflating) { - this.enqueue([this.doPong, buf, mask, toBuffer.readOnly, cb]); + this.enqueue([this.dispatch, data, false, options, cb]); } else { - this.doPong(buf, mask, toBuffer.readOnly, cb); + this.sendFrame(Sender.frame(data, options), cb); } } - /** - * Frames and sends a pong message. - * - * @param {Buffer} data The message to send - * @param {Boolean} [mask=false] Specifies whether or not to mask `data` - * @param {Boolean} [readOnly=false] Specifies whether `data` can be modified - * @param {Function} [cb] Callback - * @private - */ - doPong(data, mask, readOnly, cb) { - this.sendFrame( - Sender.frame(data, { - fin: true, - rsv1: false, - opcode: 0x0a, - mask, - maskBuffer: this._maskBuffer, - generateMask: this._generateMask, - readOnly - }), - cb - ); - } - /** * Sends a data message to the other peer. * @@ -303,11 +303,22 @@ class Sender { * @public */ send(data, options, cb) { - const buf = toBuffer(data); const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName]; let opcode = options.binary ? 2 : 1; let rsv1 = options.compress; + let byteLength; + let readOnly; + + if (typeof data === 'string') { + byteLength = Buffer.byteLength(data); + readOnly = false; + } else { + data = toBuffer(data); + byteLength = data.length; + readOnly = toBuffer.readOnly; + } + if (this._firstFragment) { this._firstFragment = false; if ( @@ -319,7 +330,7 @@ class Sender { : 'client_no_context_takeover' ] ) { - rsv1 = buf.length >= perMessageDeflate._threshold; + rsv1 = byteLength >= perMessageDeflate._threshold; } this._compress = rsv1; } else { @@ -331,30 +342,32 @@ class Sender { if (perMessageDeflate) { const opts = { + byteLength, fin: options.fin, - rsv1, - opcode, + generateMask: this._generateMask, mask: options.mask, maskBuffer: this._maskBuffer, - generateMask: this._generateMask, - readOnly: toBuffer.readOnly + opcode, + readOnly, + rsv1 }; if (this._deflating) { - this.enqueue([this.dispatch, buf, this._compress, opts, cb]); + this.enqueue([this.dispatch, data, this._compress, opts, cb]); } else { - this.dispatch(buf, this._compress, opts, cb); + this.dispatch(data, this._compress, opts, cb); } } else { this.sendFrame( - Sender.frame(buf, { + Sender.frame(data, { + byteLength, fin: options.fin, - rsv1: false, - opcode, + generateMask: this._generateMask, mask: options.mask, maskBuffer: this._maskBuffer, - generateMask: this._generateMask, - readOnly: toBuffer.readOnly + opcode, + readOnly, + rsv1: false }), cb ); @@ -362,13 +375,13 @@ class Sender { } /** - * Dispatches a data message. + * Dispatches a message. * - * @param {Buffer} data The message to send + * @param {(Buffer|String)} data The message to send * @param {Boolean} [compress=false] Specifies whether or not to compress * `data` * @param {Object} options Options object - * @param {Number} options.opcode The opcode + * @param {Number} [options.byteLength] The length in bytes of the message * @param {Boolean} [options.fin=false] Specifies whether or not to set the * FIN bit * @param {Function} [options.generateMask] The function used to generate the @@ -377,6 +390,7 @@ class Sender { * `data` * @param {Buffer} [options.maskBuffer] The buffer used to store the masking * key + * @param {Number} options.opcode The opcode * @param {Boolean} [options.readOnly=false] Specifies whether `data` can be * modified * @param {Boolean} [options.rsv1=false] Specifies whether or not to set the @@ -392,7 +406,7 @@ class Sender { const perMessageDeflate = this._extensions[PerMessageDeflate.extensionName]; - this._bufferedBytes += data.length; + this._bufferedBytes += options.byteLength; this._deflating = true; perMessageDeflate.compress(data, options.fin, (_, buf) => { if (this._socket.destroyed) { @@ -403,7 +417,8 @@ class Sender { if (typeof cb === 'function') cb(err); for (let i = 0; i < this._queue.length; i++) { - const callback = this._queue[i][4]; + const params = this._queue[i]; + const callback = params[params.length - 1]; if (typeof callback === 'function') callback(err); } @@ -411,7 +426,7 @@ class Sender { return; } - this._bufferedBytes -= data.length; + this._bufferedBytes -= options.byteLength; this._deflating = false; options.readOnly = false; this.sendFrame(Sender.frame(buf, options), cb); @@ -428,7 +443,7 @@ class Sender { while (!this._deflating && this._queue.length) { const params = this._queue.shift(); - this._bufferedBytes -= params[1].length; + this._bufferedBytes -= params[3].byteLength; Reflect.apply(params[0], this, params.slice(1)); } } @@ -440,7 +455,7 @@ class Sender { * @private */ enqueue(params) { - this._bufferedBytes += params[1].length; + this._bufferedBytes += params[3].byteLength; this._queue.push(params); } diff --git a/test/sender.test.js b/test/sender.test.js index 845ff7ad1..419a11bb0 100644 --- a/test/sender.test.js +++ b/test/sender.test.js @@ -46,6 +46,19 @@ describe('Sender', () => { assert.strictEqual(list[0][0] & 0x40, 0x40); }); + + it('accepts a string as first argument', () => { + const list = Sender.frame('€', { + readOnly: false, + rsv1: false, + mask: false, + opcode: 1, + fin: true + }); + + assert.deepStrictEqual(list[0], Buffer.from('8103', 'hex')); + assert.strictEqual(list[1], '€'); + }); }); describe('#send', () => { @@ -93,7 +106,7 @@ describe('Sender', () => { assert.strictEqual(chunks[0].length, 2); assert.notStrictEqual(chunk[0][0] & 0x40, 0x40); - assert.deepStrictEqual(chunks[1], Buffer.from('hi')); + assert.strictEqual(chunks[1], 'hi'); done(); } }); @@ -245,11 +258,12 @@ describe('Sender', () => { if (count % 2) { assert.ok(data.equals(Buffer.from([0x89, 0x02]))); - } else { + } else if (count < 8) { assert.ok(data.equals(Buffer.from([0x68, 0x69]))); + } else { + assert.strictEqual(data, 'hi'); + done(); } - - if (count === 8) done(); } }); const sender = new Sender(mockSocket, { @@ -277,11 +291,12 @@ describe('Sender', () => { if (count % 2) { assert.ok(data.equals(Buffer.from([0x8a, 0x02]))); - } else { + } else if (count < 8) { assert.ok(data.equals(Buffer.from([0x68, 0x69]))); + } else { + assert.strictEqual(data, 'hi'); + done(); } - - if (count === 8) done(); } }); const sender = new Sender(mockSocket, {