diff --git a/src/js/internal/streams/legacy.ts b/src/js/internal/streams/legacy.ts index 2ed60fff0b9c..eefd50006b48 100644 --- a/src/js/internal/streams/legacy.ts +++ b/src/js/internal/streams/legacy.ts @@ -111,8 +111,7 @@ function prependListener(emitter, event, fn) { // the prependListener() method. The goal is to eventually remove this hack. let events, existing; if (!(events = emitter._events) || !(existing = events[event])) emitter.on(event, fn); - // A fresh array, not unshift(): node:events iterates stored arrays without - // cloning, so a stored `_events` array must never be mutated in place. + // A fresh array, not unshift(): this path can't see node:events' kIterated mark, so in-place isn't safe here. else if (ArrayIsArray(existing)) events[event] = [fn, ...existing]; else events[event] = [fn, existing]; } diff --git a/src/js/node/events.ts b/src/js/node/events.ts index cca5fa632b62..054803fd39ae 100644 --- a/src/js/node/events.ts +++ b/src/js/node/events.ts @@ -39,10 +39,13 @@ const types = require("node:util/types"); let inspect: typeof import("node:util").inspect | undefined; const SymbolFor = Symbol.for; +const ArrayPrototypePop = Array.prototype.pop; const ArrayPrototypeUnshift = Array.prototype.unshift; const ReflectOwnKeys = Reflect.ownKeys; const kCapture = Symbol("kCapture"); +// Sticky mark emit() sets before iterating a listener array; on/off mutate in place unless they see it, then they install a fresh copy so the running emit() stays stable. +const kIterated = Symbol("kIterated"); // Set when `_events` was preallocated (streams do this): removeListener then // writes `undefined` instead of `delete`, keeping one shared JSC Structure // so the (StructureID, name)-keyed megamorphic cache stays hot. @@ -141,14 +144,13 @@ function emitError(emitter, args) { throw err; // Unhandled 'error' event } -// A listener list is a bare function for a single listener, else an array -// (like node). Arrays are never mutated in place - mutators install a copy - -// so a stored list can be iterated with no defensive clone. +// A listener list is a bare function for a single listener, else an array (as in node). function applyHandlers(handlers, emitter, args) { if (typeof handlers === "function") { handlers.$apply(emitter, args); return; } + if (!handlers[kIterated]) handlers[kIterated] = true; for (let i = 0, { length } = handlers; i < length; i++) { handlers[i].$apply(emitter, args); } @@ -208,9 +210,8 @@ const emitWithoutRejectionCapture = function emit(type, ...args) { } return true; } - // No defensive clone: stored arrays are never mutated in place (mutators - // install a copy), so this list stays stable for the whole loop even if a - // listener adds/removes listeners. + // kIterated diverts reentrant on/off to a copy, so `handler` stays stable. + if (!handler[kIterated]) handler[kIterated] = true; for (let i = 0, { length } = handler; i < length; i++) { const listener = handler[i]; switch (args.length) { @@ -268,9 +269,8 @@ const emitWithRejectionCapture = function emit(type, ...args) { } return true; } - // No defensive clone: stored arrays are never mutated in place (mutators - // install a copy), so this list stays stable for the whole loop even if a - // listener adds/removes listeners. + // kIterated diverts reentrant on/off to a copy, so `handler` stays stable. + if (!handler[kIterated]) handler[kIterated] = true; for (let i = 0, { length } = handler; i < length; i++) { const listener = handler[i]; let result; @@ -322,8 +322,12 @@ function _addListener(target, type, fn, prepend) { var handlers; if (typeof existing === "function") { handlers = events[type] = prepend ? [fn, existing] : [existing, fn]; - } else { + } else if (existing[kIterated]) { handlers = events[type] = copyWithInserted(existing, fn, prepend); + } else { + if (prepend) ArrayPrototypeUnshift.$call(existing, fn); + else $arrayPush(existing, fn); + handlers = existing; } var m = _getMaxListeners(target); if (m > 0 && handlers.length > m && !handlers.warned) { @@ -343,9 +347,7 @@ EventEmitterPrototype.prependListener = function prependListener(type, fn) { return this; }; -// Copy-on-write: emit iterates stored arrays with no clone, so new listeners -// land in a fresh array; `warned` carries over so the leak warning fires once. -// An inline loop beats concat/slice here ~10x (host-call boundary). +// Fresh-array insert for a kIterated `list`; propagates `warned`, inline loop beats concat/slice ~10x. function copyWithInserted(list, fn, prepend) { const n = list.length; const copy = $newArrayWithSize(n + 1); @@ -447,21 +449,31 @@ EventEmitterPrototype.removeListener = function removeListener(type, listener) { } if (position < 0) return this; - // Copy-remove (arrays are never mutated in place), and store a lone - // survivor bare like node does, so `_events[type]` shape matches theirs. - const n = list.length; - const copy = $newArrayWithSize(n - 1); - for (let i = 0, j = 0; i < n; i++) { - if (i !== position) copy[j++] = list[i]; + if (list[kIterated]) { + const n = list.length; + const copy = $newArrayWithSize(n - 1); + for (let i = 0, j = 0; i < n; i++) { + if (i !== position) copy[j++] = list[i]; + } + if (list.warned) copy.warned = true; + events[type] = copy.length === 1 ? copy[0] : copy; + } else if (list.length === 2) { + events[type] = list[1 - position]; + } else { + spliceOne(list, position); } - if (list.warned) copy.warned = true; - events[type] = copy.length === 1 ? copy[0] : copy; if (events.removeListener !== undefined) this.emit("removeListener", type, listener.listener ?? listener); return this; }; +// Node's internal/util spliceOne: O(1) at the tail, so a tail-first drain is linear. +function spliceOne(list, index) { + for (; index + 1 < list.length; index++) list[index] = list[index + 1]; + ArrayPrototypePop.$call(list); +} + EventEmitterPrototype.off = EventEmitterPrototype.removeListener; EventEmitterPrototype.removeAllListeners = function removeAllListeners(type) { @@ -498,8 +510,7 @@ EventEmitterPrototype.removeAllListeners = function removeAllListeners(type) { if (typeof listeners === "function") { this.removeListener(type, listeners); } else if (listeners !== undefined) { - // LIFO order. `listeners` is our own snapshot; each removeListener call - // installs a fresh array (or bare fn / nothing), so it stays intact here. + // LIFO, as in node; `listeners` either shrinks from the tail in step with i, or (if kIterated) is COW'd away and stays intact. for (let i = listeners.length - 1; i >= 0; i--) this.removeListener(type, listeners[i]); } return this; diff --git a/test/js/node/events/event-emitter.test.ts b/test/js/node/events/event-emitter.test.ts index f95a0464cdc1..1839bbe68f09 100644 --- a/test/js/node/events/event-emitter.test.ts +++ b/test/js/node/events/event-emitter.test.ts @@ -264,6 +264,69 @@ describe("EventEmitter", () => { expect(EventEmitter.prototype.removeListener).toBe(EventEmitter.prototype.off); }); + // Node mutates the stored listener array in place (push / spliceOne), so + // capturing `_events[type]` and then adding or removing keeps the same array + // reference. A copy-on-write scheme allocates a fresh array per call instead, + // turning N adds (or a tail-first drain of N listeners) into O(N^2) work. + // https://github.com/oven-sh/bun/issues/3770 + // https://github.com/oven-sh/bun/issues/3734 + test("on/removeListener mutate the stored listener array in place", () => { + const ee = new EventEmitter() as any; + ee.setMaxListeners(0); + const f1 = () => {}; + const f2 = () => {}; + const f3 = () => {}; + const f4 = () => {}; + ee.on("x", f1); + ee.on("x", f2); + const list = ee._events.x; + expect(Array.isArray(list)).toBe(true); + + ee.on("x", f3); + expect(ee._events.x).toBe(list); + expect(list).toEqual([f1, f2, f3]); + + ee.prependListener("x", f4); + expect(ee._events.x).toBe(list); + expect(list).toEqual([f4, f1, f2, f3]); + + ee.removeListener("x", f3); + expect(ee._events.x).toBe(list); + expect(list).toEqual([f4, f1, f2]); + + ee.removeListener("x", f4); + expect(ee._events.x).toBe(list); + expect(list).toEqual([f1, f2]); + }); + + // The in-place mutation above is only safe because emit() marks the array it + // is iterating: a listener that adds or removes on the same event gets a + // fresh copy, so the running loop still sees its original snapshot. This + // also has to hold across a nested emit() of the same event, which must not + // clear the outer emit()'s mark when it finishes. + test("nested emit + removeListener inside a listener preserves the outer emit snapshot", () => { + const ee = new EventEmitter(); + const calls: string[] = []; + let depth = 0; + const a = () => calls.push("a" + depth); + const b = () => { + calls.push("b" + depth); + if (depth === 0) { + depth = 1; + ee.emit("x"); + depth = 0; + ee.removeListener("x", c); + } + }; + const c = () => calls.push("c" + depth); + ee.on("x", a); + ee.on("x", b); + ee.on("x", c); + ee.emit("x"); + expect(calls).toEqual(["a0", "b0", "a1", "b1", "c1", "c0"]); + expect(ee.listeners("x")).toEqual([a, b]); + }); + test("prependListener", () => { const myEmitter = new EventEmitter(); const order: number[] = [];