|
| 1 | +// Flags: --expose-gc |
| 2 | +'use strict'; |
| 3 | + |
| 4 | +const common = require('../common'); |
| 5 | +if (!common.hasCrypto) |
| 6 | + common.skip('missing crypto'); |
| 7 | +const assert = require('assert'); |
| 8 | +const http2 = require('http2'); |
| 9 | +const net = require('net'); |
| 10 | + |
| 11 | +const FRAME_DATA = 0; |
| 12 | +const FRAME_HEADERS = 1; |
| 13 | +const FRAME_RST_STREAM = 3; |
| 14 | +const FRAME_SETTINGS = 4; |
| 15 | +const FRAME_WINDOW_UPDATE = 8; |
| 16 | +const FLAG_END_STREAM = 1; |
| 17 | +const FLAG_END_HEADERS = 4; |
| 18 | +const FLAG_PRIORITY = 0x20; |
| 19 | +const INITIAL_WINDOW_SIZE = 65_535; |
| 20 | +const FIRST_WINDOW_SIZE = 3_000_000; |
| 21 | +const SECOND_WINDOW_INCREMENT = 20_000_000; |
| 22 | +const LARGE_RESPONSE_SIZE = 30_000_000; |
| 23 | +const SMALL_FIRST_BUFFER_SIZE = 80 * 1024; |
| 24 | +const SMALL_SECOND_BUFFER_SIZE = 20 * 1024; |
| 25 | +const SMALL_RESPONSE_SIZE = |
| 26 | + SMALL_FIRST_BUFFER_SIZE + SMALL_SECOND_BUFFER_SIZE; |
| 27 | +const EXPECTED_SMALL_PAYLOAD_SIZE = INITIAL_WINDOW_SIZE; |
| 28 | + |
| 29 | +const preface = Buffer.from('PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n'); |
| 30 | + |
| 31 | +function frame(type, flags, streamID, payload = Buffer.alloc(0)) { |
| 32 | + const header = Buffer.alloc(9); |
| 33 | + header.writeUIntBE(payload.length, 0, 3); |
| 34 | + header[3] = type; |
| 35 | + header[4] = flags; |
| 36 | + header.writeUInt32BE(streamID, 5); |
| 37 | + return Buffer.concat([header, payload]); |
| 38 | +} |
| 39 | + |
| 40 | +function windowUpdate(streamID, increment) { |
| 41 | + const payload = Buffer.alloc(4); |
| 42 | + payload.writeUInt32BE(increment); |
| 43 | + return frame(FRAME_WINDOW_UPDATE, 0, streamID, payload); |
| 44 | +} |
| 45 | + |
| 46 | +function request(path, streamID, priority = false) { |
| 47 | + const authority = Buffer.from('127.0.0.1'); |
| 48 | + let payload = Buffer.concat([ |
| 49 | + Buffer.from([ |
| 50 | + 0x82, // :method: GET |
| 51 | + 0x86, // :scheme: http |
| 52 | + 0x01, authority.length, // Literal :authority |
| 53 | + ]), |
| 54 | + authority, |
| 55 | + Buffer.from([0x04, path.length]), // Literal :path |
| 56 | + Buffer.from(path), |
| 57 | + ]); |
| 58 | + let flags = FLAG_END_STREAM | FLAG_END_HEADERS; |
| 59 | + |
| 60 | + if (priority) { |
| 61 | + const priorityPayload = Buffer.alloc(5); |
| 62 | + // Make this stream exclusively dependent on stream 1 so its DATA is |
| 63 | + // serialized after stream 1's DATA in the pending socket write. |
| 64 | + priorityPayload.writeUInt32BE(0x80000001); |
| 65 | + priorityPayload[4] = 15; |
| 66 | + payload = Buffer.concat([priorityPayload, payload]); |
| 67 | + flags |= FLAG_PRIORITY; |
| 68 | + } |
| 69 | + |
| 70 | + return frame(FRAME_HEADERS, flags, streamID, payload); |
| 71 | +} |
| 72 | + |
| 73 | +const replacementBuffers = []; |
| 74 | +let resumeClient; |
| 75 | + |
| 76 | +const server = http2.createServer({ maxSessionMemory: 64 }); |
| 77 | +server.on('stream', common.mustCall((stream, headers) => { |
| 78 | + const path = headers[':path']; |
| 79 | + stream.on('error', () => {}); |
| 80 | + stream.respond(); |
| 81 | + |
| 82 | + if (path === '/A') { |
| 83 | + stream.end(Buffer.alloc(LARGE_RESPONSE_SIZE, 0x41)); |
| 84 | + return; |
| 85 | + } |
| 86 | + |
| 87 | + stream.on('close', common.mustCall(() => { |
| 88 | + setImmediate(() => { |
| 89 | + // Collect the canceled write and reuse allocations of the same size. |
| 90 | + // The pending socket slices must continue to refer to the original data. |
| 91 | + globalThis.gc(); |
| 92 | + replacementBuffers.push(...Array.from( |
| 93 | + { length: 32 }, |
| 94 | + () => Buffer.alloc(SMALL_FIRST_BUFFER_SIZE, 0x56))); |
| 95 | + // Only now let the backpressured socket write drain. |
| 96 | + resumeClient(); |
| 97 | + }); |
| 98 | + })); |
| 99 | + // Use _writev() so the sliced first buffer does not carry the request's |
| 100 | + // completion reference. Every buffer must retain a separate owner. |
| 101 | + stream.cork(); |
| 102 | + stream.write(Buffer.alloc(SMALL_FIRST_BUFFER_SIZE, 0x53)); |
| 103 | + stream.end(Buffer.alloc(SMALL_SECOND_BUFFER_SIZE, 0x53)); |
| 104 | +}, 2)); |
| 105 | + |
| 106 | +server.listen(0, common.mustCall(() => { |
| 107 | + const socket = net.connect(server.address().port, common.localhostIPv4); |
| 108 | + let received = Buffer.alloc(0); |
| 109 | + let stream1Received = 0; |
| 110 | + let secondWindowSent = false; |
| 111 | + let resetSent = false; |
| 112 | + const stream3Chunks = []; |
| 113 | + |
| 114 | + resumeClient = () => socket.resume(); |
| 115 | + |
| 116 | + const finish = common.mustCall((payload) => { |
| 117 | + assert.strictEqual(payload.length, EXPECTED_SMALL_PAYLOAD_SIZE); |
| 118 | + assert.ok( |
| 119 | + payload.equals(Buffer.alloc(EXPECTED_SMALL_PAYLOAD_SIZE, 0x53)), |
| 120 | + 'pending DATA used memory from a canceled write'); |
| 121 | + |
| 122 | + socket.destroy(); |
| 123 | + server.close(common.mustCall()); |
| 124 | + }); |
| 125 | + |
| 126 | + function onStream1Data() { |
| 127 | + if (!secondWindowSent) { |
| 128 | + if (stream1Received < FIRST_WINDOW_SIZE) |
| 129 | + return; |
| 130 | + // Stream 1 has consumed its whole first window, so the server has no |
| 131 | + // DATA left to send. Open stream 3 and the windows together: the server |
| 132 | + // packs the remaining stream 1 DATA and stream 3 DATA (serialized after |
| 133 | + // stream 1) into a single large socket write. |
| 134 | + secondWindowSent = true; |
| 135 | + socket.write(Buffer.concat([ |
| 136 | + request('/S', 3, true), |
| 137 | + windowUpdate(0, SECOND_WINDOW_INCREMENT + SMALL_RESPONSE_SIZE), |
| 138 | + windowUpdate(1, SECOND_WINDOW_INCREMENT), |
| 139 | + ])); |
| 140 | + return; |
| 141 | + } |
| 142 | + |
| 143 | + if (resetSent || stream1Received <= FIRST_WINDOW_SIZE) |
| 144 | + return; |
| 145 | + |
| 146 | + // Receiving DATA beyond the first window proves the large write has been |
| 147 | + // issued, and stream 3 DATA was packed into it. Stop reading so that write |
| 148 | + // stays pending, then reset stream 3 while its slices are still queued. |
| 149 | + // The server resumes us once the stream is closed and memory is reused. |
| 150 | + resetSent = true; |
| 151 | + socket.pause(); |
| 152 | + const payload = Buffer.alloc(4); |
| 153 | + payload.writeUInt32BE(http2.constants.NGHTTP2_CANCEL); |
| 154 | + socket.write(frame(FRAME_RST_STREAM, 0, 3, payload)); |
| 155 | + } |
| 156 | + |
| 157 | + function parseFrames() { |
| 158 | + let offset = 0; |
| 159 | + while (offset + 9 <= received.length) { |
| 160 | + const length = received.readUIntBE(offset, 3); |
| 161 | + if (offset + 9 + length > received.length) |
| 162 | + break; |
| 163 | + |
| 164 | + const type = received[offset + 3]; |
| 165 | + const streamID = received.readUInt32BE(offset + 5) & 0x7fffffff; |
| 166 | + if (type === FRAME_DATA && streamID === 3) { |
| 167 | + stream3Chunks.push(received.subarray( |
| 168 | + offset + 9, |
| 169 | + offset + 9 + length)); |
| 170 | + } else if (type === FRAME_DATA && streamID === 1) { |
| 171 | + stream1Received += length; |
| 172 | + onStream1Data(); |
| 173 | + } |
| 174 | + offset += 9 + length; |
| 175 | + } |
| 176 | + received = received.subarray(offset); |
| 177 | + |
| 178 | + const payload = Buffer.concat(stream3Chunks); |
| 179 | + if (payload.length < EXPECTED_SMALL_PAYLOAD_SIZE) |
| 180 | + return; |
| 181 | + |
| 182 | + finish(payload); |
| 183 | + } |
| 184 | + |
| 185 | + socket.on('connect', () => { |
| 186 | + socket.write(Buffer.concat([ |
| 187 | + preface, |
| 188 | + frame(FRAME_SETTINGS, 0, 0), |
| 189 | + request('/A', 1), |
| 190 | + windowUpdate(0, FIRST_WINDOW_SIZE - INITIAL_WINDOW_SIZE), |
| 191 | + windowUpdate(1, FIRST_WINDOW_SIZE - INITIAL_WINDOW_SIZE), |
| 192 | + ])); |
| 193 | + }); |
| 194 | + |
| 195 | + socket.on('data', (chunk) => { |
| 196 | + received = Buffer.concat([received, chunk]); |
| 197 | + parseFrames(); |
| 198 | + }); |
| 199 | +})); |
0 commit comments