Commit cad6d99
child_process: speed up 'advanced' ipc receiving
PR-URL: #42931
Reviewed-By: Ruben Bridgewater <ruben@bridgewater.de>
Reviewed-By: Luigi Pinca <luigipinca@gmail.com>
Reviewed-By: Zeyu "Alex" Yang <himself65@outlook.com>1 parent a9ab41c commit cad6d99
1 file changed
Lines changed: 33 additions & 11 deletions
| Original file line number | Diff line number | Diff line change | |
|---|---|---|---|
@@ -4,6 +4,8 @@ const { | |||
| 4 | 4 | JSONParse, | |
| 5 | 5 | JSONStringify, | |
| 6 | 6 | StringPrototypeSplit, | |
| 7 | + ArrayPrototypePush, | ||
| 8 | + ReflectApply, | ||
| 7 | 9 | Symbol, | |
| 8 | 10 | TypedArrayPrototypeSubarray, | |
| 9 | 11 | } = primordials; | |
@@ -13,8 +15,10 @@ const v8 = require('v8'); | |||
| 13 | 15 | const { isArrayBufferView } = require('internal/util/types'); | |
| 14 | 16 | const assert = require('internal/assert'); | |
| 15 | 17 | const { streamBaseState, kLastWriteWasAsync } = internalBinding('stream_wrap'); | |
| 18 | + const { readUInt32BE } = require('internal/buffer'); | ||
| 16 | 19 | ||
| 17 | 20 | const kMessageBuffer = Symbol('kMessageBuffer'); | |
| 21 | + const kMessageBufferSize = Symbol('kMessageBufferSize'); | ||
| 18 | 22 | const kJSONBuffer = Symbol('kJSONBuffer'); | |
| 19 | 23 | const kStringDecoder = Symbol('kStringDecoder'); | |
| 20 | 24 | ||
@@ -51,29 +55,47 @@ class ChildProcessDeserializer extends v8.DefaultDeserializer { | |||
| 51 | 55 | // (aka 'advanced') | |
| 52 | 56 | const advanced = { | |
| 53 | 57 | initMessageChannel(channel) { | |
| 54 | - channel[kMessageBuffer] = Buffer.alloc(0); | ||
| 58 | + channel[kMessageBuffer] = []; | ||
| 59 | + channel[kMessageBufferSize] = 0; | ||
| 55 | 60 | channel.buffering = false; | |
| 56 | 61 | }, | |
| 57 | 62 | ||
| 58 | 63 | *parseChannelMessages(channel, readData) { | |
| 59 | 64 | if (readData.length === 0) return; | |
| 60 | 65 | ||
| 61 | - let messageBuffer = Buffer.concat([channel[kMessageBuffer], readData]); | ||
| 62 | - while (messageBuffer.length > 4) { | ||
| 63 | - const size = messageBuffer.readUInt32BE(); | ||
| 64 | - if (messageBuffer.length < 4 + size) { | ||
| 65 | - break; | ||
| 66 | - } | ||
| 66 | + ArrayPrototypePush(channel[kMessageBuffer], readData); | ||
| 67 | + channel[kMessageBufferSize] += readData.length; | ||
| 68 | + | ||
| 69 | + // Index 0 should always be present because we just pushed data into it. | ||
| 70 | + let messageBufferHead = channel[kMessageBuffer][0]; | ||
| 71 | + while (messageBufferHead.length >= 4) { | ||
| 72 | + // We call `readUInt32BE` manually here, because this is faster than first converting | ||
| 73 | + // it to a buffer and using `readUInt32BE` on that. | ||
| 74 | + const fullMessageSize = ReflectApply(readUInt32BE, messageBufferHead, [0]) + 4; | ||
| 75 | + | ||
| 76 | + if (channel[kMessageBufferSize] < fullMessageSize) break; | ||
| 77 | + | ||
| 78 | + const concatenatedBuffer = channel[kMessageBuffer].length === 1 ? | ||
| 79 | + channel[kMessageBuffer][0] : | ||
| 80 | + Buffer.concat( | ||
| 81 | + channel[kMessageBuffer], | ||
| 82 | + channel[kMessageBufferSize] | ||
| 83 | + ); | ||
| 67 | 84 | ||
| 68 | 85 | const deserializer = new ChildProcessDeserializer( | |
| 69 | - TypedArrayPrototypeSubarray(messageBuffer, 4, 4 + size)); | ||
| 70 | - messageBuffer = TypedArrayPrototypeSubarray(messageBuffer, 4 + size); | ||
| 86 | + TypedArrayPrototypeSubarray(concatenatedBuffer, 4, fullMessageSize) | ||
| 87 | + ); | ||
| 88 | + | ||
| 89 | + messageBufferHead = TypedArrayPrototypeSubarray(concatenatedBuffer, fullMessageSize); | ||
| 90 | + channel[kMessageBufferSize] = messageBufferHead.length; | ||
| 91 | + channel[kMessageBuffer] = | ||
| 92 | + channel[kMessageBufferSize] !== 0 ? [messageBufferHead] : []; | ||
| 71 | 93 | ||
| 72 | 94 | deserializer.readHeader(); | |
| 73 | 95 | yield deserializer.readValue(); | |
| 74 | 96 | } | |
| 75 | - channel[kMessageBuffer] = messageBuffer; | ||
| 76 | - channel.buffering = messageBuffer.length > 0; | ||
| 97 | + | ||
| 98 | + channel.buffering = channel[kMessageBufferSize] > 0; | ||
| 77 | 99 | }, | |
| 78 | 100 | ||
| 79 | 101 | writeChannelMessage(channel, req, message, handle) { | |
0 commit comments