From 288372ee71188ed1b880ece906619cae9fc7d33d Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Mon, 1 Jun 2026 21:45:49 +0200 Subject: [PATCH] Fix NodeAv memory leak (fixes #392) --- packages/server/src/audio-decoder.ts | 72 +++++++---- packages/server/src/audio-encoder.ts | 186 +++++++++++++++------------ packages/server/src/index.ts | 4 + packages/server/src/video-decoder.ts | 89 ++++++++----- packages/server/src/video-encoder.ts | 166 +++++++++++++----------- test/node/server-extension.test.ts | 82 ++++++++++++ 6 files changed, 390 insertions(+), 209 deletions(-) diff --git a/packages/server/src/audio-decoder.ts b/packages/server/src/audio-decoder.ts index 4d94a42..ed9faf4 100644 --- a/packages/server/src/audio-decoder.ts +++ b/packages/server/src/audio-decoder.ts @@ -12,10 +12,28 @@ import { CODEC_TO_CODEC_ID, getChannelLayout } from './misc'; import { assert, toUint8Array } from '../../../src/misc'; import { AvFrameAudioSampleResource } from './audio-sample'; +type NodeAvState = { + frame: NodeAv.Frame; + packet: NodeAv.Packet; + codecContext: NodeAv.CodecContext | null; +}; + +const freeState = (state: NodeAvState) => { + state.codecContext?.freeContext(); + state.frame.free(); + state.packet.free(); +}; + +// Needed for proper freeing if close isn't called +let finalizationRegistry: FinalizationRegistry | null = null; +if (typeof FinalizationRegistry !== 'undefined') { + finalizationRegistry = new FinalizationRegistry((state) => { + freeState(state); + }); +} + export class NodeAvAudioDecoder extends CustomAudioDecoder { - frame!: NodeAv.Frame; - packet!: NodeAv.Packet; - codecContext!: NodeAv.CodecContext; + state!: NodeAvState; // eslint-disable-next-line @typescript-eslint/no-unused-vars static override supports(codec: AudioCodec, config: AudioDecoderConfig): boolean { @@ -29,10 +47,13 @@ export class NodeAvAudioDecoder extends CustomAudioDecoder { } async init(): Promise { - this.frame = new NodeAv.Frame(); - this.frame.alloc(); - this.packet = new NodeAv.Packet(); - this.packet.alloc(); + const frame = new NodeAv.Frame(); + frame.alloc(); + const packet = new NodeAv.Packet(); + packet.alloc(); + this.state = { frame, packet, codecContext: null }; + + finalizationRegistry?.register(this, this.state, this); const codecId = CODEC_TO_CODEC_ID[this.codec]; assert(codecId !== undefined); @@ -57,22 +78,26 @@ export class NodeAvAudioDecoder extends CustomAudioDecoder { const ret = await codecContext.open2(); NodeAv.FFmpegError.throwIfError(ret, 'Open codec context'); - this.codecContext = codecContext; + this.state.codecContext = codecContext; } async decode(packet: EncodedPacket): Promise { - this.packet.isKeyframe = packet.type === 'key'; - this.packet.data = Buffer.from(packet.data); - this.packet.timeBase = { num: 1, den: this.config.sampleRate }; - this.packet.pts = BigInt(Math.round(packet.timestamp * this.config.sampleRate)); - this.packet.dts = NodeAv.AV_NOPTS_VALUE; - this.packet.duration = BigInt(Math.round(packet.duration * this.config.sampleRate)); + assert(this.state.codecContext); - const ret = await this.codecContext.sendPacket(this.packet); + this.state.packet.isKeyframe = packet.type === 'key'; + this.state.packet.data = Buffer.from(packet.data); + this.state.packet.timeBase = { num: 1, den: this.config.sampleRate }; + this.state.packet.pts = BigInt(Math.round(packet.timestamp * this.config.sampleRate)); + this.state.packet.dts = NodeAv.AV_NOPTS_VALUE; + this.state.packet.duration = BigInt(Math.round(packet.duration * this.config.sampleRate)); + + const ret = await this.state.codecContext.sendPacket(this.state.packet); NodeAv.FFmpegError.throwIfError(ret, 'Send packet'); + this.state.packet.unref(); // Don't need the data again, so just unref it + while (true) { - const receiveRet = await this.codecContext.receiveFrame(this.frame); + const receiveRet = await this.state.codecContext.receiveFrame(this.state.frame); if (receiveRet === NodeAv.AVERROR_EAGAIN || receiveRet === NodeAv.AVERROR_EOF) { break; } @@ -84,7 +109,7 @@ export class NodeAvAudioDecoder extends CustomAudioDecoder { receiveFrame(ret: number) { NodeAv.FFmpegError.throwIfError(ret, 'Receive frame'); - const clone = this.frame.clone(); + const clone = this.state.frame.clone(); if (!clone) { throw new Error('Allocation failure during frame clone.'); } @@ -94,13 +119,15 @@ export class NodeAvAudioDecoder extends CustomAudioDecoder { } async flush(): Promise { + assert(this.state.codecContext); + // Send null packet to signal flush - const ret = await this.codecContext.sendPacket(null); + const ret = await this.state.codecContext.sendPacket(null); NodeAv.FFmpegError.throwIfError(ret, 'Flush decoder'); // Keep receiving frames until no more are available while (true) { - const receiveRet = await this.codecContext.receiveFrame(this.frame); + const receiveRet = await this.state.codecContext.receiveFrame(this.state.frame); if (receiveRet === NodeAv.AVERROR_EAGAIN || receiveRet === NodeAv.AVERROR_EOF) { // No more frames available break; @@ -109,12 +136,11 @@ export class NodeAvAudioDecoder extends CustomAudioDecoder { this.receiveFrame(receiveRet); } - this.codecContext.flushBuffers(); + this.state.codecContext.flushBuffers(); } close(): MaybePromise { - this.codecContext.freeContext(); - this.frame.free(); - this.packet.free(); + finalizationRegistry?.unregister(this); + freeState(this.state); } } diff --git a/packages/server/src/audio-encoder.ts b/packages/server/src/audio-encoder.ts index 7b9dd16..cdb2186 100644 --- a/packages/server/src/audio-encoder.ts +++ b/packages/server/src/audio-encoder.ts @@ -32,19 +32,39 @@ const AC3_SAMPLE_RATES = [32000, 44100, 48000]; const FRAME_SIZE_FALLBACK = 1024; // Just 'cause +type NodeAvState = { + frame: NodeAv.Frame; + packet: NodeAv.Packet; + codecContext: NodeAv.CodecContext | null; + resampler: NodeAv.SoftwareResampleContext | null; + dstFrame: NodeAv.Frame | null; +}; + +const freeState = (state: NodeAvState) => { + state.codecContext?.freeContext(); + state.frame.free(); + state.packet.free(); + state.dstFrame?.free(); + state.resampler?.free(); +}; + +// Needed for proper freeing if close isn't called +let finalizationRegistry: FinalizationRegistry | null = null; +if (typeof FinalizationRegistry !== 'undefined') { + finalizationRegistry = new FinalizationRegistry((state) => { + freeState(state); + }); +} + export class NodeAvAudioEncoder extends CustomAudioEncoder { - frame!: NodeAv.Frame; - packet!: NodeAv.Packet; + state!: NodeAvState; avCodec!: NodeAv.Codec; - codecContext: NodeAv.CodecContext | null = null; firstExpectedTimestamp: number | null = null; outputTimestampOffset = 0; - resampler: NodeAv.SoftwareResampleContext | null = null; inputParametersKey: string | null = null; resamplerInputSampleRate: number | null = null; nextResamplerPts: bigint | null = null; - dstFrame: NodeAv.Frame | null = null; packetEmitted = false; adtsHeaderTemplate: AdtsHeaderTemplate | null = null; @@ -76,10 +96,13 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder { } async init(): Promise { - this.frame = new NodeAv.Frame(); - this.frame.alloc(); - this.packet = new NodeAv.Packet(); - this.packet.alloc(); + const frame = new NodeAv.Frame(); + frame.alloc(); + const packet = new NodeAv.Packet(); + packet.alloc(); + this.state = { frame, packet, codecContext: null, resampler: null, dstFrame: null }; + + finalizationRegistry?.register(this, this.state, this); const codecId = CODEC_TO_CODEC_ID[this.codec]; assert(codecId !== undefined); @@ -95,7 +118,7 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder { } async createCodecContext() { - assert(this.codecContext === null); + assert(this.state.codecContext === null); const codecContext = new NodeAv.CodecContext(); codecContext.allocContext3(this.avCodec); @@ -122,28 +145,32 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder { const ret = await codecContext.open2(); NodeAv.FFmpegError.throwIfError(ret, 'Open codec context'); - this.codecContext = codecContext; + this.state.codecContext = codecContext; } async encode(audioSample: AudioSample): Promise { - if (this.codecContext === null) { + if (this.state.codecContext === null) { await this.createCodecContext(); - assert(this.codecContext); + assert(this.state.codecContext); } this.firstExpectedTimestamp ??= audioSample.timestamp; if (audioSample._data instanceof AvFrameAudioSampleResource) { - this.frame.ref(audioSample._data.frame); + // Release any buffers still referenced from the previous encode before reffing the new frame, otherwise + // av_frame_ref leaks them + // https://github.com/Vanilagy/mediabunny/issues/392 + this.state.frame.unref(); + this.state.frame.ref(audioSample._data.frame); } else { - copyAudioSampleToAvFrame(audioSample, this.frame); + copyAudioSampleToAvFrame(audioSample, this.state.frame); } - this.frame.pts = BigInt(Math.round(audioSample.timestamp * this.config.sampleRate)); - this.frame.duration = BigInt(Math.round(audioSample.duration * this.config.sampleRate)); - this.frame.timeBase = new NodeAv.Rational(1, this.config.sampleRate); + this.state.frame.pts = BigInt(Math.round(audioSample.timestamp * this.config.sampleRate)); + this.state.frame.duration = BigInt(Math.round(audioSample.duration * this.config.sampleRate)); + this.state.frame.timeBase = new NodeAv.Rational(1, this.config.sampleRate); - const key = `${this.frame.sampleRate}:${this.frame.channels}:${this.frame.format}`; + const key = `${this.state.frame.sampleRate}:${this.state.frame.channels}:${this.state.frame.format}`; if (this.inputParametersKey !== null && this.inputParametersKey !== key) { throw new Error( 'Input audio parameters changed. For this audio encoder, you cannot change the input audio' @@ -156,83 +183,83 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder { // 1. Format conversion is needed (sample format, sample rate, or channel count differs) // 2. The codec requires fixed frame sizes const requiresResampler - = this.codecContext.frameSize > 0 - || this.codecContext.sampleFormat !== this.frame.format - || this.codecContext.sampleRate !== this.frame.sampleRate - || this.codecContext.channels !== this.frame.channels; + = this.state.codecContext.frameSize > 0 + || this.state.codecContext.sampleFormat !== this.state.frame.format + || this.state.codecContext.sampleRate !== this.state.frame.sampleRate + || this.state.codecContext.channels !== this.state.frame.channels; if (requiresResampler) { - if (!this.resampler) { - this.resampler = new NodeAv.SoftwareResampleContext(); - this.resamplerInputSampleRate = this.frame.sampleRate; + if (!this.state.resampler) { + this.state.resampler = new NodeAv.SoftwareResampleContext(); + this.resamplerInputSampleRate = this.state.frame.sampleRate; - const outLayout = getChannelLayout(this.codecContext.channels); - const inLayout = getChannelLayout(this.frame.channels); + const outLayout = getChannelLayout(this.state.codecContext.channels); + const inLayout = getChannelLayout(this.state.frame.channels); - const ret = this.resampler.allocSetOpts2( - outLayout, this.codecContext.sampleFormat, this.codecContext.sampleRate, - inLayout, this.frame.format as NodeAv.AVSampleFormat, this.frame.sampleRate, + const ret = this.state.resampler.allocSetOpts2( + outLayout, this.state.codecContext.sampleFormat, this.state.codecContext.sampleRate, + inLayout, this.state.frame.format as NodeAv.AVSampleFormat, this.state.frame.sampleRate, ); NodeAv.FFmpegError.throwIfError(ret, 'allocSetOpts2'); - const ret2 = this.resampler.init(); + const ret2 = this.state.resampler.init(); NodeAv.FFmpegError.throwIfError(ret2, 'init'); - this.dstFrame = new NodeAv.Frame(); - this.dstFrame.alloc(); - this.dstFrame.channelLayout = outLayout; - this.dstFrame.sampleRate = this.codecContext.sampleRate; - this.dstFrame.format = this.codecContext.sampleFormat; - this.dstFrame.nbSamples = this.codecContext.frameSize || FRAME_SIZE_FALLBACK; - this.dstFrame.duration = BigInt(this.dstFrame.nbSamples); - this.dstFrame.allocBuffer(); + this.state.dstFrame = new NodeAv.Frame(); + this.state.dstFrame.alloc(); + this.state.dstFrame.channelLayout = outLayout; + this.state.dstFrame.sampleRate = this.state.codecContext.sampleRate; + this.state.dstFrame.format = this.state.codecContext.sampleFormat; + this.state.dstFrame.nbSamples = this.state.codecContext.frameSize || FRAME_SIZE_FALLBACK; + this.state.dstFrame.duration = BigInt(this.state.dstFrame.nbSamples); + this.state.dstFrame.allocBuffer(); - this.nextResamplerPts = this.frame.pts; + this.nextResamplerPts = this.state.frame.pts; } - const inputBuffers = this.frame.data; + const inputBuffers = this.state.frame.data; if (!inputBuffers) { throw new DOMException('Frame has no data', 'EncodingError'); } - await this.resampler.convert(null, 0, inputBuffers, this.frame.nbSamples); + await this.state.resampler.convert(null, 0, inputBuffers, this.state.frame.nbSamples); await this.pullResampledFrames(); } else { - await this.sendFrameAndReceivePackets(this.frame); + await this.sendFrameAndReceivePackets(this.state.frame); } } async pullResampledFrames() { - assert(this.codecContext); - assert(this.resampler); - assert(this.dstFrame); + assert(this.state.codecContext); + assert(this.state.resampler); + assert(this.state.dstFrame); assert(this.nextResamplerPts !== null); - const frameSize = this.codecContext.frameSize || FRAME_SIZE_FALLBACK; + const frameSize = this.state.codecContext.frameSize || FRAME_SIZE_FALLBACK; while (true) { - const available = this.resampler.getOutSamples(0); + const available = this.state.resampler.getOutSamples(0); if (available < frameSize) { break; } - await this.resampler.convert(this.dstFrame.data, frameSize, null, 0); + await this.state.resampler.convert(this.state.dstFrame.data, frameSize, null, 0); - this.dstFrame.pts = this.nextResamplerPts; + this.state.dstFrame.pts = this.nextResamplerPts; - await this.sendFrameAndReceivePackets(this.dstFrame); + await this.sendFrameAndReceivePackets(this.state.dstFrame); this.nextResamplerPts += BigInt(frameSize); } } async sendFrameAndReceivePackets(frame: NodeAv.Frame | null) { - assert(this.codecContext); + assert(this.state.codecContext); - const ret = await this.codecContext.sendFrame(frame); + const ret = await this.state.codecContext.sendFrame(frame); NodeAv.FFmpegError.throwIfError(ret, 'Send frame'); while (true) { - const receiveRet = await this.codecContext.receivePacket(this.packet); + const receiveRet = await this.state.codecContext.receivePacket(this.state.packet); if (receiveRet === NodeAv.AVERROR_EAGAIN || receiveRet === NodeAv.AVERROR_EOF) { break; } @@ -242,18 +269,18 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder { } receivePacket(ret: number) { - assert(this.codecContext); + assert(this.state.codecContext); assert(this.firstExpectedTimestamp !== null); NodeAv.FFmpegError.throwIfError(ret, 'Receive packet'); - if (!this.packet.data) { + if (!this.state.packet.data) { return; } - let timestamp = Number(this.packet.pts) / this.codecContext.sampleRate; - const duration = Number(this.packet.duration) / this.codecContext.sampleRate; + let timestamp = Number(this.state.packet.pts) / this.state.codecContext.sampleRate; + const duration = Number(this.state.packet.duration) / this.state.codecContext.sampleRate; - let data: Uint8Array = this.packet.data; + let data: Uint8Array = this.state.packet.data; let metadata: EncodedAudioChunkMetadata | undefined; if (this.packetEmitted) { @@ -265,8 +292,8 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder { this.outputTimestampOffset = Math.max(this.firstExpectedTimestamp - timestamp, 0); const codecString = this.config.codec; - let description = this.codecContext.extraData - ? toUint8Array(this.codecContext.extraData) + let description = this.state.codecContext.extraData + ? toUint8Array(this.state.codecContext.extraData) : undefined; if (this.codec === 'aac') { @@ -311,8 +338,8 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder { metadata = { decoderConfig: { codec: codecString, - sampleRate: this.codecContext.sampleRate, - numberOfChannels: this.codecContext.channels, + sampleRate: this.state.codecContext.sampleRate, + numberOfChannels: this.state.codecContext.channels, description, }, }; @@ -344,53 +371,50 @@ export class NodeAvAudioEncoder extends CustomAudioEncoder { } async flush(): Promise { - if (!this.codecContext) { + if (!this.state.codecContext) { return; } outer: - if (this.resampler) { + if (this.state.resampler) { assert(this.resamplerInputSampleRate !== null); - const currentOutSamples = this.resampler.getOutSamples(0); + const currentOutSamples = this.state.resampler.getOutSamples(0); if (currentOutSamples === 0) { break outer; // Clean cut-off point } - const frameSize = this.codecContext.frameSize || FRAME_SIZE_FALLBACK; + const frameSize = this.state.codecContext.frameSize || FRAME_SIZE_FALLBACK; assert(currentOutSamples < frameSize); // Because if it's more, it would've already been retrieved const inputSamplesNeeded = Math.ceil( - ((frameSize - currentOutSamples) / this.codecContext.sampleRate) * this.resamplerInputSampleRate, + ((frameSize - currentOutSamples) / this.state.codecContext.sampleRate) * this.resamplerInputSampleRate, ); - this.resampler.injectSilence(inputSamplesNeeded); + this.state.resampler.injectSilence(inputSamplesNeeded); await this.pullResampledFrames(); } await this.sendFrameAndReceivePackets(null); - this.codecContext.freeContext(); - this.codecContext = null; + this.state.codecContext.freeContext(); + this.state.codecContext = null; this.packetEmitted = false; this.firstExpectedTimestamp = null; this.outputTimestampOffset = 0; this.adtsHeaderTemplate = null; - this.resampler?.free(); - this.resampler = null; + this.state.resampler?.free(); + this.state.resampler = null; this.inputParametersKey = null; this.resamplerInputSampleRate = null; this.nextResamplerPts = null; - this.dstFrame?.free(); - this.dstFrame = null; + this.state.dstFrame?.free(); + this.state.dstFrame = null; } close(): MaybePromise { - this.codecContext?.freeContext(); - this.frame.free(); - this.packet.free(); - this.dstFrame?.free(); - this.resampler?.free(); + finalizationRegistry?.unregister(this); + freeState(this.state); } } diff --git a/packages/server/src/index.ts b/packages/server/src/index.ts index a59337c..002ea2f 100644 --- a/packages/server/src/index.ts +++ b/packages/server/src/index.ts @@ -78,6 +78,8 @@ export { AvFrameAudioSampleResource } from './audio-sample'; export const toAvFrame = async (sample: VideoSample | AudioSample, frame: NodeAv.Frame) => { if (sample instanceof VideoSample) { if (sample._data instanceof AvFrameVideoSampleResource) { + // We're overriding the frame, so release whatever it referenced before, otherwise av_frame_ref leaks it + frame.unref(); frame.ref(sample._data.frame); } else { if (sample.format === null) { @@ -92,6 +94,8 @@ export const toAvFrame = async (sample: VideoSample | AudioSample, frame: NodeAv frame.timeBase = new NodeAv.Rational(1, 1e6); } else { if (sample._data instanceof AvFrameAudioSampleResource) { + // We're overriding the frame, so release whatever it referenced before, otherwise av_frame_ref leaks it + frame.unref(); frame.ref(sample._data.frame); } else { copyAudioSampleToAvFrame(sample, frame); diff --git a/packages/server/src/video-decoder.ts b/packages/server/src/video-decoder.ts index 7b56827..bbddef1 100644 --- a/packages/server/src/video-decoder.ts +++ b/packages/server/src/video-decoder.ts @@ -12,10 +12,28 @@ import { CODEC_TO_CODEC_ID, getHardwareDecoderCodec } from './misc'; import { assert, binarySearchLessOrEqual, simplifyRational, toUint8Array } from '../../../src/misc'; import { AvFrameVideoSampleResource } from './video-sample'; +type NodeAvState = { + frame: NodeAv.Frame; + packet: NodeAv.Packet; + codecContext: NodeAv.CodecContext | null; +}; + +const freeState = (state: NodeAvState) => { + state.codecContext?.freeContext(); + state.frame.free(); + state.packet.free(); +}; + +// Needed for proper freeing if close isn't called +let finalizationRegistry: FinalizationRegistry | null = null; +if (typeof FinalizationRegistry !== 'undefined') { + finalizationRegistry = new FinalizationRegistry((state) => { + freeState(state); + }); +} + export class NodeAvVideoDecoder extends CustomVideoDecoder { - frame!: NodeAv.Frame; - packet!: NodeAv.Packet; - codecContext: NodeAv.CodecContext | null = null; + state!: NodeAvState; pixelAspectRatio!: Rational; // Bookkeeping to restore the original timing information @@ -33,14 +51,17 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder { } async init() { - this.frame = new NodeAv.Frame(); - this.frame.alloc(); - this.packet = new NodeAv.Packet(); - this.packet.alloc(); + const frame = new NodeAv.Frame(); + frame.alloc(); + const packet = new NodeAv.Packet(); + packet.alloc(); + this.state = { frame, packet, codecContext: null }; + + finalizationRegistry?.register(this, this.state, this); } async initCodecContext(packet: EncodedPacket) { - assert(this.codecContext === null); + assert(this.state.codecContext === null); const codecId = CODEC_TO_CODEC_ID[this.codec]; assert(codecId !== undefined); @@ -85,28 +106,29 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder { const ret = await codecContext.open2(); NodeAv.FFmpegError.throwIfError(ret, 'Open codec context'); - this.codecContext = codecContext; + this.state.codecContext = codecContext; } async decode(packet: EncodedPacket) { - if (this.codecContext === null) { + if (this.state.codecContext === null) { await this.initCodecContext(packet); - assert(this.codecContext); } - this.packet.isKeyframe = packet.type === 'key'; - this.packet.data = Buffer.from(packet.data); - this.packet.timeBase = { num: 1, den: 1e6 }; - this.packet.pts = BigInt(packet.microsecondTimestamp); - this.packet.dts = NodeAv.AV_NOPTS_VALUE; - this.packet.duration = BigInt(packet.microsecondDuration); + assert(this.state.codecContext); + + this.state.packet.isKeyframe = packet.type === 'key'; + this.state.packet.data = Buffer.from(packet.data); + this.state.packet.timeBase = { num: 1, den: 1e6 }; + this.state.packet.pts = BigInt(packet.microsecondTimestamp); + this.state.packet.dts = NodeAv.AV_NOPTS_VALUE; + this.state.packet.duration = BigInt(packet.microsecondDuration); if (packet.sideData.alpha) { const matroskaBlockAdditional = Buffer.alloc(8 + packet.sideData.alpha.byteLength); matroskaBlockAdditional[7] = 1; // BlockAddId matroskaBlockAdditional.set(packet.sideData.alpha, 8); - this.packet.addSideData(NodeAv.AV_PKT_DATA_MATROSKA_BLOCKADDITIONAL, matroskaBlockAdditional); + this.state.packet.addSideData(NodeAv.AV_PKT_DATA_MATROSKA_BLOCKADDITIONAL, matroskaBlockAdditional); } const preciseTimingIndex = binarySearchLessOrEqual( @@ -141,11 +163,13 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder { } } - const ret = await this.codecContext.sendPacket(this.packet); + const ret = await this.state.codecContext.sendPacket(this.state.packet); NodeAv.FFmpegError.throwIfError(ret, 'Send packet'); + this.state.packet.unref(); // Don't need the data again, so just unref it + while (true) { - const receiveRet = await this.codecContext.receiveFrame(this.frame); + const receiveRet = await this.state.codecContext.receiveFrame(this.state.frame); if (receiveRet === NodeAv.AVERROR_EAGAIN || receiveRet === NodeAv.AVERROR_EOF) { break; } @@ -157,14 +181,14 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder { receiveFrame(ret: number) { NodeAv.FFmpegError.throwIfError(ret, 'Receive frame'); - this.frame.sampleAspectRatio = new NodeAv.Rational(this.pixelAspectRatio.num, this.pixelAspectRatio.den); + this.state.frame.sampleAspectRatio = new NodeAv.Rational(this.pixelAspectRatio.num, this.pixelAspectRatio.den); - let timestamp = Number(this.frame.pts) / 1e6; - let duration = Number(this.frame.duration) / 1e6; + let timestamp = Number(this.state.frame.pts) / 1e6; + let duration = Number(this.state.frame.duration) / 1e6; const preciseTimingIndex = binarySearchLessOrEqual( this.preciseTimings, - Number(this.frame.pts), + Number(this.state.frame.pts), x => x.microsecondTimestamp, ); const entry = preciseTimingIndex !== -1 @@ -173,7 +197,7 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder { // If there's a relevant timing entry, refine the frame's timing data to get better accuracy than // microseconds - if (entry && entry.microsecondTimestamp === Number(this.frame.pts)) { + if (entry && entry.microsecondTimestamp === Number(this.state.frame.pts)) { if (entry.timestampIsValid) { timestamp = entry.timestamp; } @@ -182,7 +206,7 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder { } } - const clone = this.frame.clone(); + const clone = this.state.frame.clone(); if (!clone) { throw new Error('Frame clone allocation failed.'); } @@ -194,17 +218,17 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder { } async flush() { - if (!this.codecContext) { + if (!this.state.codecContext) { return; } // Send null packet to signal flush - const ret = await this.codecContext.sendPacket(null); + const ret = await this.state.codecContext.sendPacket(null); NodeAv.FFmpegError.throwIfError(ret, 'Flush decoder'); // Keep receiving frames until no more are available while (true) { - const receiveRet = await this.codecContext.receiveFrame(this.frame); + const receiveRet = await this.state.codecContext.receiveFrame(this.state.frame); if (receiveRet === NodeAv.AVERROR_EAGAIN || receiveRet === NodeAv.AVERROR_EOF) { // No more frames available break; @@ -213,12 +237,11 @@ export class NodeAvVideoDecoder extends CustomVideoDecoder { this.receiveFrame(receiveRet); } - this.codecContext.flushBuffers(); + this.state.codecContext.flushBuffers(); } close(): MaybePromise { - this.codecContext?.freeContext(); - this.frame.free(); - this.packet.free(); + finalizationRegistry?.unregister(this); + freeState(this.state); } } diff --git a/packages/server/src/video-encoder.ts b/packages/server/src/video-encoder.ts index f6ed3a0..5bea44d 100644 --- a/packages/server/src/video-encoder.ts +++ b/packages/server/src/video-encoder.ts @@ -41,17 +41,36 @@ import { import { extractVideoCodecString } from '../../../src/codec'; import { assert, binarySearchLessOrEqual, simplifyRational, toUint8Array } from '../../../src/misc'; +type NodeAvState = { + frame: NodeAv.Frame; + packet: NodeAv.Packet; + codecContext: NodeAv.CodecContext | null; + scaler: NodeAv.SoftwareScaleContext | null; + dstFrame: NodeAv.Frame | null; +}; + +const freeState = (state: NodeAvState) => { + state.codecContext?.freeContext(); + state.frame.free(); + state.packet.free(); + state.scaler?.freeContext(); + state.dstFrame?.free(); +}; + +// Needed for proper freeing if close isn't called +let finalizationRegistry: FinalizationRegistry | null = null; +if (typeof FinalizationRegistry !== 'undefined') { + finalizationRegistry = new FinalizationRegistry((state) => { + freeState(state); + }); +} + export class NodeAvVideoEncoder extends CustomVideoEncoder { - frame!: NodeAv.Frame; - packet!: NodeAv.Packet; + state!: NodeAvState; avCodec!: NodeAv.Codec; - codecContext: NodeAv.CodecContext | null = null; lastBuffer: Buffer | null = null; packetEmitted = false; - - scaler: NodeAv.SoftwareScaleContext | null = null; lastScalerKey: string | null = null; - dstFrame: NodeAv.Frame | null = null; // Bookkeeping to restore the original timing information preciseTimings: { @@ -68,12 +87,16 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder { } async init(): Promise { - this.frame = new NodeAv.Frame(); - this.frame.alloc(); - this.frame.timeBase = new NodeAv.Rational(1, 1e6); + const frame = new NodeAv.Frame(); + frame.alloc(); + frame.timeBase = new NodeAv.Rational(1, 1e6); - this.packet = new NodeAv.Packet(); - this.packet.alloc(); + const packet = new NodeAv.Packet(); + packet.alloc(); + + this.state = { frame, packet, codecContext: null, scaler: null, dstFrame: null }; + + finalizationRegistry?.register(this, this.state, this); const codecId = CODEC_TO_CODEC_ID[this.codec]; assert(codecId !== undefined); @@ -97,7 +120,7 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder { } async createCodecContext() { - assert(this.codecContext === null); + assert(this.state.codecContext === null); const codecContext = new NodeAv.CodecContext(); codecContext.allocContext3(this.avCodec); @@ -177,65 +200,69 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder { const ret = await codecContext.open2(); NodeAv.FFmpegError.throwIfError(ret, 'Open codec context'); - this.codecContext = codecContext; + this.state.codecContext = codecContext; } async encode(videoSample: VideoSample, options: VideoEncoderEncodeOptions): Promise { - if (this.codecContext === null) { + if (this.state.codecContext === null) { await this.createCodecContext(); - assert(this.codecContext); } + assert(this.state.codecContext); if (videoSample._data instanceof AvFrameVideoSampleResource) { - this.frame.ref(videoSample._data.frame); + // Release any buffers still referenced from the previous encode before reffing the new frame, otherwise + // av_frame_ref leaks them + // https://github.com/Vanilagy/mediabunny/issues/392 + this.state.frame.unref(); + this.state.frame.ref(videoSample._data.frame); } else { if (videoSample.format === null) { throw new Error('Cannot encode foreign VideoSample with unknown (null) format.'); } - this.lastBuffer = await copyVideoSampleToAvFrame(videoSample, this.frame, this.lastBuffer); + this.lastBuffer = await copyVideoSampleToAvFrame(videoSample, this.state.frame, this.lastBuffer); } - let frameToEncode = this.frame; + let frameToEncode = this.state.frame; const requiresScaler - = this.codecContext.pixelFormat !== this.frame.format - || this.codecContext.width !== this.frame.width - || this.codecContext.height !== this.frame.height; + = this.state.codecContext.pixelFormat !== this.state.frame.format + || this.state.codecContext.width !== this.state.frame.width + || this.state.codecContext.height !== this.state.frame.height; if (requiresScaler) { - if (!this.scaler) { - this.scaler = new NodeAv.SoftwareScaleContext(); + if (!this.state.scaler) { + this.state.scaler = new NodeAv.SoftwareScaleContext(); } - const key = `${this.frame.width}x${this.frame.height}:${this.frame.format}`; + const key = `${this.state.frame.width}x${this.state.frame.height}:${this.state.frame.format}`; const needsConfigure = key !== this.lastScalerKey; if (needsConfigure) { - this.scaler.getContext( - this.frame.width, this.frame.height, this.frame.format as NodeAv.AVPixelFormat, - this.codecContext.width, this.codecContext.height, this.codecContext.pixelFormat, + this.state.scaler.getContext( + this.state.frame.width, this.state.frame.height, this.state.frame.format as NodeAv.AVPixelFormat, + this.state.codecContext.width, this.state.codecContext.height, this.state.codecContext.pixelFormat, NodeAv.SWS_FAST_BILINEAR, ); this.lastScalerKey = key; - const ret = this.scaler.initContext(); + const ret = this.state.scaler.initContext(); NodeAv.FFmpegError.throwIfError(ret, 'initContext'); } - if (!this.dstFrame) { - this.dstFrame = new NodeAv.Frame(); - this.dstFrame.alloc(); - this.dstFrame.width = this.codecContext.width; - this.dstFrame.height = this.codecContext.height; - this.dstFrame.format = this.codecContext.pixelFormat; - this.dstFrame.allocBuffer(); + if (!this.state.dstFrame) { + this.state.dstFrame = new NodeAv.Frame(); + this.state.dstFrame.alloc(); + this.state.dstFrame.width = this.state.codecContext.width; + this.state.dstFrame.height = this.state.codecContext.height; + this.state.dstFrame.format = this.state.codecContext.pixelFormat; + this.state.dstFrame.allocBuffer(); } - await this.scaler.scaleFrame(this.dstFrame, this.frame); - this.dstFrame.copyProps(this.frame); - frameToEncode = this.dstFrame; + await this.state.scaler.scaleFrame(this.state.dstFrame, this.state.frame); + this.state.dstFrame.copyProps(this.state.frame); + frameToEncode = this.state.dstFrame; } frameToEncode.pts = BigInt(videoSample.microsecondTimestamp); @@ -282,12 +309,12 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder { } } - const ret = await this.codecContext.sendFrame(frameToEncode); + const ret = await this.state.codecContext.sendFrame(frameToEncode); NodeAv.FFmpegError.throwIfError(ret, 'Send frame'); // Keep receiving packets until no more are available for this frame while (true) { - const receiveRet = await this.codecContext.receivePacket(this.packet); + const receiveRet = await this.state.codecContext.receivePacket(this.state.packet); if (receiveRet === NodeAv.AVERROR_EAGAIN || receiveRet === NodeAv.AVERROR_EOF) { break; } @@ -297,20 +324,20 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder { } receivePacket(ret: number) { - assert(this.codecContext); + assert(this.state.codecContext); NodeAv.FFmpegError.throwIfError(ret, 'Receive packet'); - if (!this.packet.data) { + if (!this.state.packet.data) { return; } - let packetData = toUint8Array(this.packet.data); + let packetData = toUint8Array(this.state.packet.data); - let timestamp = Number(this.packet.pts) / 1e6; - let duration = Number(this.packet.duration) / 1e6; + let timestamp = Number(this.state.packet.pts) / 1e6; + let duration = Number(this.state.packet.duration) / 1e6; const preciseTimingIndex = binarySearchLessOrEqual( this.preciseTimings, - Number(this.packet.pts), + Number(this.state.packet.pts), x => x.microsecondTimestamp, ); const entry = preciseTimingIndex !== -1 @@ -319,7 +346,7 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder { // If there's a relevant timing entry, refine the packet's timing data to get better accuracy than // microseconds - if (entry && entry.microsecondTimestamp === Number(this.packet.pts)) { + if (entry && entry.microsecondTimestamp === Number(this.state.packet.pts)) { if (entry.timestampIsValid) { timestamp = entry.timestamp; } @@ -346,14 +373,14 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder { let serializedRecord: Uint8Array; if (this.codec === 'avc') { - const record = extractAvcDecoderConfigurationRecord(this.packet.data); + const record = extractAvcDecoderConfigurationRecord(this.state.packet.data); if (!record) { throw new Error('Invalid AVC data, could not extract decoder configuration record.'); } serializedRecord = serializeAvcDecoderConfigurationRecord(record); } else { - const record = extractHevcDecoderConfigurationRecord(this.packet.data); + const record = extractHevcDecoderConfigurationRecord(this.state.packet.data); if (!record) { throw new Error('Invalid HEVC data, could not extract decoder configuration record.'); } @@ -488,14 +515,14 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder { } const sideData: EncodedPacketSideData = {}; - const matroskaBlockAdditional = this.packet.getSideData(NodeAv.AV_PKT_DATA_MATROSKA_BLOCKADDITIONAL); + const matroskaBlockAdditional = this.state.packet.getSideData(NodeAv.AV_PKT_DATA_MATROSKA_BLOCKADDITIONAL); if (matroskaBlockAdditional) { sideData.alpha = toUint8Array(matroskaBlockAdditional).subarray(8); // Skip the BlockAddId } const packet = new EncodedPacket( packetData, - this.packet.isKeyframe ? 'key' : 'delta', + this.state.packet.isKeyframe ? 'key' : 'delta', timestamp, duration, undefined, @@ -507,21 +534,19 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder { // Create the decoder config metadata.decoderConfig = { codec: decoderConfigCodecString, - codedWidth: this.codecContext.width, - codedHeight: this.codecContext.height, - displayAspectWidth: this.config.displayWidth ?? this.codecContext.width, - displayAspectHeight: this.config.displayHeight ?? this.codecContext.height, + codedWidth: this.state.codecContext.width, + codedHeight: this.state.codecContext.height, + displayAspectWidth: this.config.displayWidth ?? this.state.codecContext.width, + displayAspectHeight: this.config.displayHeight ?? this.state.codecContext.height, description: decoderConfigDescription ?? undefined, colorSpace: { - primaries: - unmapColorPrimaries(this.codecContext.colorPrimaries) as VideoColorPrimaries, - matrix: - unmapMatrixCoefficients(this.codecContext.colorSpace) as VideoMatrixCoefficients, + primaries: unmapColorPrimaries(this.state.codecContext.colorPrimaries) as VideoColorPrimaries, + matrix: unmapMatrixCoefficients(this.state.codecContext.colorSpace) as VideoMatrixCoefficients, transfer: - unmapTransferCharacteristics(this.codecContext.colorTrc) as VideoTransferCharacteristics, - fullRange: this.codecContext.colorRange === NodeAv.AVCOL_RANGE_JPEG + unmapTransferCharacteristics(this.state.codecContext.colorTrc) as VideoTransferCharacteristics, + fullRange: this.state.codecContext.colorRange === NodeAv.AVCOL_RANGE_JPEG ? true - : this.codecContext.colorRange === NodeAv.AVCOL_RANGE_MPEG + : this.state.codecContext.colorRange === NodeAv.AVCOL_RANGE_MPEG ? false : undefined, }, @@ -533,14 +558,14 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder { } async flush(): Promise { - if (this.codecContext) { + if (this.state.codecContext) { // Send null frame to signal flush - const ret = await this.codecContext.sendFrame(null); + const ret = await this.state.codecContext.sendFrame(null); NodeAv.FFmpegError.throwIfError(ret, 'Send frame'); // Keep receiving packets until no more are available while (true) { - const receiveRet = await this.codecContext.receivePacket(this.packet); + const receiveRet = await this.state.codecContext.receivePacket(this.state.packet); if (receiveRet === NodeAv.AVERROR_EAGAIN || receiveRet === NodeAv.AVERROR_EOF) { break; } @@ -548,8 +573,8 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder { this.receivePacket(receiveRet); } - this.codecContext.freeContext(); - this.codecContext = null; + this.state.codecContext.freeContext(); + this.state.codecContext = null; // The codec is done now and can't be reused. Any subsequent encode call will first need to recreate a // codec context. } @@ -558,10 +583,7 @@ export class NodeAvVideoEncoder extends CustomVideoEncoder { } close(): MaybePromise { - this.codecContext?.freeContext(); - this.frame.free(); - this.packet.free(); - this.scaler?.freeContext(); - this.dstFrame?.free(); + finalizationRegistry?.unregister(this); + freeState(this.state); } } diff --git a/test/node/server-extension.test.ts b/test/node/server-extension.test.ts index 916d94b..72e6009 100644 --- a/test/node/server-extension.test.ts +++ b/test/node/server-extension.test.ts @@ -871,6 +871,88 @@ describe('Video', async () => { expect(plane[3]).toBe(0xff); }); + // https://github.com/Vanilagy/mediabunny/issues/392 + test('Memory doesn\'t leak', async () => { + const encoder = new NodeAvVideoEncoder(); + // @ts-expect-error Readonly + encoder.codec = 'avc'; + // @ts-expect-error Readonly + encoder.config = { + codec: buildVideoCodecString('avc', 1920, 1080, 1e6), + width: 1920, + height: 1080, + bitrate: 1e6, + } satisfies VideoEncoderConfig; + + const pendingPackets: EncodedPacket[] = []; + const decodedFrames: VideoSample[] = []; + let decoderConfig: VideoDecoderConfig | null = null; + + // @ts-expect-error Readonly + encoder.onPacket = (packet: EncodedPacket, meta: EncodedVideoChunkMetadata) => { + if (meta.decoderConfig) { + decoderConfig = meta.decoderConfig; + } + pendingPackets.push(packet); + }; + + await encoder.init(); + + // Encode a single white frame to get one packet (and the decoder config) to drive the loop with + const data = new Uint8Array(1920 * 1080 * 4).fill(0xff); // White + using bootstrapSample = new VideoSample(data, { + format: 'RGBX', + codedWidth: 1920, + codedHeight: 1080, + timestamp: 0, + duration: 1 / 30, + }); + await encoder.encode(bootstrapSample, {}); + await encoder.flush(); + assert(decoderConfig); + assert(pendingPackets.length > 0); + + const sourcePacket = pendingPackets[0]!; + pendingPackets.length = 0; + + const decoder = new NodeAvVideoDecoder(); + // @ts-expect-error Readonly + decoder.codec = 'avc'; + // @ts-expect-error Readonly + decoder.config = decoderConfig; + // @ts-expect-error Readonly + decoder.onSample = (sample: VideoSample) => { + decodedFrames.push(sample); + }; + await decoder.init(); + + const iterations = 200; + let baselineRss = 0; + + for (let i = 1; i <= iterations; i++) { + await decoder.decode(sourcePacket.clone({ timestamp: i / 30 })); + + for (const frame of decodedFrames) { + await encoder.encode(frame, {}); + frame.close(); + } + decodedFrames.length = 0; + pendingPackets.length = 0; + + if (i === 1) { + baselineRss = process.memoryUsage().rss; + } + } + + const finalRss = process.memoryUsage().rss; + const growth = finalRss - baselineRss; + + expect(growth).toBeLessThan(100 * 1024 * 1024); + + await encoder.close(); + await decoder.close(); + }); + describe('VideoSample transformation', () => { // 400x400 image: red everywhere, with a 200x200 blue square filling the bottom-left quadrant. const TEST_IMAGE = (() => {