Skip to content

Commit 63b7794

Browse files
mcollinaUzlopak
mcollina
and
Uzlopak
authored
use FinalizationRegistry to cancel the body if response is collected (#3199)
* use FinalizationRegistry to cancel the body if response is collected Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * fixup Signed-off-by: Matteo Collina <[email protected]> * Update lib/dispatcher/client-h2.js Co-authored-by: Aras Abbasi <[email protected]> --------- Signed-off-by: Matteo Collina <[email protected]> Co-authored-by: Aras Abbasi <[email protected]>
1 parent c0dc3dd commit 63b7794

4 files changed

Lines changed: 98 additions & 10 deletions

File tree

‎lib/web/fetch/index.js‎

Lines changed: 25 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -120,12 +120,16 @@ class Fetch extends EE {
120120
}
121121
}
122122

123+
function handleFetchDone (response) {
124+
finalizeAndReportTiming(response, 'fetch')
125+
}
126+
123127
// https://fetch.spec.whatwg.org/#fetch-method
124128
function fetch (input, init = undefined) {
125129
webidl.argumentLengthCheck(arguments, 1, 'globalThis.fetch')
126130

127131
// 1. Let p be a new promise.
128-
const p = createDeferredPromise()
132+
let p = createDeferredPromise()
129133

130134
// 2. Let requestObject be the result of invoking the initial value of
131135
// Request as constructor with input and init as arguments. If this throws
@@ -185,16 +189,17 @@ function fetch (input, init = undefined) {
185189
// 3. Abort controller with requestObject’s signal’s abort reason.
186190
controller.abort(requestObject.signal.reason)
187191

192+
const realResponse = responseObject?.deref()
193+
188194
// 4. Abort the fetch() call with p, request, responseObject,
189195
// and requestObject’s signal’s abort reason.
190-
abortFetch(p, request, responseObject, requestObject.signal.reason)
196+
abortFetch(p, request, realResponse, requestObject.signal.reason)
191197
}
192198
)
193199

194200
// 12. Let handleFetchDone given response response be to finalize and
195201
// report timing with response, globalObject, and "fetch".
196-
const handleFetchDone = (response) =>
197-
finalizeAndReportTiming(response, 'fetch')
202+
// see function handleFetchDone
198203

199204
// 13. Set controller to the result of calling fetch given request,
200205
// with processResponseEndOfBody set to handleFetchDone, and processResponse
@@ -228,10 +233,11 @@ function fetch (input, init = undefined) {
228233

229234
// 4. Set responseObject to the result of creating a Response object,
230235
// given response, "immutable", and relevantRealm.
231-
responseObject = fromInnerResponse(response, 'immutable')
236+
responseObject = new WeakRef(fromInnerResponse(response, 'immutable'))
232237

233238
// 5. Resolve p with responseObject.
234-
p.resolve(responseObject)
239+
p.resolve(responseObject.deref())
240+
p = null
235241
}
236242

237243
controller = fetching({
@@ -314,7 +320,10 @@ const markResourceTiming = performance.markResourceTiming
314320
// https://fetch.spec.whatwg.org/#abort-fetch
315321
function abortFetch (p, request, responseObject, error) {
316322
// 1. Reject promise with error.
317-
p.reject(error)
323+
if (p) {
324+
// We might have already resolved the promise at this stage
325+
p.reject(error)
326+
}
318327

319328
// 2. If request’s body is not null and is readable, then cancel request’s
320329
// body with error.
@@ -1066,7 +1075,10 @@ function fetchFinale (fetchParams, response) {
10661075
// 4. If fetchParams’s process response is non-null, then queue a fetch task to run fetchParams’s
10671076
// process response given response, with fetchParams’s task destination.
10681077
if (fetchParams.processResponse != null) {
1069-
queueMicrotask(() => fetchParams.processResponse(response))
1078+
queueMicrotask(() => {
1079+
fetchParams.processResponse(response)
1080+
fetchParams.processResponse = null
1081+
})
10701082
}
10711083

10721084
// 5. Let internalResponse be response, if response is a network error; otherwise response’s internal response.
@@ -1884,7 +1896,11 @@ async function httpNetworkFetch (
18841896
// 12. Let cancelAlgorithm be an algorithm that aborts fetchParams’s
18851897
// controller with reason, given reason.
18861898
const cancelAlgorithm = (reason) => {
1887-
fetchParams.controller.abort(reason)
1899+
// If the aborted fetch was already terminated, then we do not
1900+
// need to do anything.
1901+
if (!isCancelled(fetchParams)) {
1902+
fetchParams.controller.abort(reason)
1903+
}
18881904
}
18891905

18901906
// 13. Let highWaterMark be a non-negative, non-NaN number, chosen by

‎lib/web/fetch/response.js‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,9 +26,23 @@ const { URLSerializer } = require('./data-url')
2626
const { kHeadersList, kConstruct } = require('../../core/symbols')
2727
const assert = require('node:assert')
2828
const { types } = require('node:util')
29+
const { isDisturbed, isErrored } = require('node:stream')
2930

3031
const textEncoder = new TextEncoder('utf-8')
3132

33+
const hasFinalizationRegistry = globalThis.FinalizationRegistry && process.version.indexOf('v18') !== 0
34+
let registry
35+
36+
if (hasFinalizationRegistry) {
37+
registry = new FinalizationRegistry((stream) => {
38+
if (!stream.locked && !isDisturbed(stream) && !isErrored(stream)) {
39+
stream.cancel('Response object has been garbage collected').catch(noop)
40+
}
41+
})
42+
}
43+
44+
function noop () {}
45+
3246
// https://fetch.spec.whatwg.org/#response-class
3347
class Response {
3448
// Creates network error Response.
@@ -510,6 +524,11 @@ function fromInnerResponse (innerResponse, guard) {
510524
response[kHeaders] = new Headers(kConstruct)
511525
response[kHeaders][kHeadersList] = innerResponse.headersList
512526
response[kHeaders][kGuard] = guard
527+
528+
if (hasFinalizationRegistry && innerResponse.body?.stream) {
529+
registry.register(response, innerResponse.body.stream)
530+
}
531+
513532
return response
514533
}
515534

‎package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -77,7 +77,7 @@
7777
"test:eventsource:nobuild": "borp --expose-gc -p \"test/eventsource/*.js\"",
7878
"test:fuzzing": "node test/fuzzing/fuzzing.test.js",
7979
"test:fetch": "npm run build:node && npm run test:fetch:nobuild",
80-
"test:fetch:nobuild": "borp --expose-gc -p \"test/fetch/*.js\" && npm run test:webidl && npm run test:busboy",
80+
"test:fetch:nobuild": "borp --timeout 180000 --expose-gc --concurrency 1 -p \"test/fetch/*.js\" && npm run test:webidl && npm run test:busboy",
8181
"test:interceptors": "borp -p \"test/interceptors/*.js\"",
8282
"test:jest": "cross-env NODE_V8_COVERAGE= jest",
8383
"test:unit": "borp --expose-gc -p \"test/*.js\"",

‎test/fetch/fire-and-forget.js‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,53 @@
1+
'use strict'
2+
3+
const { randomFillSync } = require('node:crypto')
4+
const { setTimeout: sleep } = require('timers/promises')
5+
const { test } = require('node:test')
6+
const { fetch, Agent, setGlobalDispatcher } = require('../..')
7+
const { createServer } = require('node:http')
8+
const { closeServerAsPromise } = require('../utils/node-http')
9+
10+
const blob = randomFillSync(new Uint8Array(1024 * 512))
11+
12+
// Enable when/if FinalizationRegistry in Node.js 18 becomes stable again
13+
const isNode18 = process.version.startsWith('v18')
14+
15+
test('does not need the body to be consumed to continue', { timeout: 180_000, skip: isNode18 }, async (t) => {
16+
const agent = new Agent({
17+
keepAliveMaxTimeout: 10,
18+
keepAliveTimeoutThreshold: 10
19+
})
20+
setGlobalDispatcher(agent)
21+
const server = createServer((req, res) => {
22+
res.writeHead(200)
23+
res.end(blob)
24+
})
25+
t.after(closeServerAsPromise(server))
26+
27+
await new Promise((resolve) => {
28+
server.listen(0, resolve)
29+
})
30+
31+
const url = new URL(`http://127.0.0.1:${server.address().port}`)
32+
33+
const batch = 50
34+
const delay = 0
35+
let total = 0
36+
while (total < 10000) {
37+
// eslint-disable-next-line no-undef
38+
gc(true)
39+
const array = new Array(batch)
40+
for (let i = 0; i < batch; i++) {
41+
array[i] = fetch(url).catch(() => {})
42+
}
43+
await Promise.all(array)
44+
await sleep(delay)
45+
46+
console.log(
47+
'RSS',
48+
(process.memoryUsage.rss() / 1024 / 1024) | 0,
49+
'MB after',
50+
(total += batch) + ' fetch() requests'
51+
)
52+
}
53+
})

0 commit comments

Comments
 (0)