Fully update MP3 demuxer to use new packet reading system & mutex, add MP3 reading tests

This commit is contained in:
Vanilagy
2025-12-21 11:52:54 +01:00
parent ece27cca79
commit 00efb69a7d
6 changed files with 287 additions and 141 deletions
+11 -1
View File
@@ -17,8 +17,18 @@ export abstract class Demuxer {
this.input = input;
}
abstract computeDuration(): Promise<number>;
abstract getTracks(): Promise<InputTrack[]>;
abstract getMimeType(): Promise<string>;
abstract getMetadataTags(): Promise<MetadataTags>;
async computeDuration(): Promise<number> {
const tracks = await this.getTracks();
if (tracks.length === 0) {
return 0;
}
// eslint-disable-next-line @typescript-eslint/await-thenable
const trackDurations = await Promise.all(tracks.map(x => x.computeDuration()));
return Math.max(...trackDurations);
}
}
+32 -10
View File
@@ -38,14 +38,12 @@ export interface InputTrackBacking {
getLanguageCode(): string;
getTimeResolution(): number;
getDisposition(): TrackDisposition;
getFirstTimestamp(): Promise<number>;
computeDuration(): Promise<number>;
getFirstPacket(res: ResultValue<EncodedPacket | null>, options: PacketRetrievalOptions): MaybePromise<Yo>;
getNextPacket(res: ResultValue<EncodedPacket | null>, packet: EncodedPacket, options: PacketRetrievalOptions): MaybePromise<Yo>;
getPacket(res: ResultValue<EncodedPacket | null>, timestamp: number, options: PacketRetrievalOptions): MaybePromise<Yo>;
getKeyPacket(res: ResultValue<EncodedPacket | null>, timestamp: number, options: PacketRetrievalOptions): MaybePromise<Yo>;
getNextKeyPacket(res: ResultValue<EncodedPacket | null>, packet: EncodedPacket, options: PacketRetrievalOptions): MaybePromise<Yo>;
getFirstPacket(res: ResultValue<EncodedPacket | null>, options: PacketRetrievalOptions): Promise<Yo>;
getNextPacket(res: ResultValue<EncodedPacket | null>, packet: EncodedPacket, options: PacketRetrievalOptions): Promise<Yo>;
getPacket(res: ResultValue<EncodedPacket | null>, timestamp: number, options: PacketRetrievalOptions): Promise<Yo>;
getKeyPacket(res: ResultValue<EncodedPacket | null>, timestamp: number, options: PacketRetrievalOptions): Promise<Yo>;
getNextKeyPacket(res: ResultValue<EncodedPacket | null>, packet: EncodedPacket, options: PacketRetrievalOptions): Promise<Yo>;
// getFirstPacket(options: PacketRetrievalOptions): Promise<EncodedPacket | null>;
// getPacket(timestamp: number, options: PacketRetrievalOptions): Promise<EncodedPacket | null>;
@@ -147,12 +145,36 @@ export abstract class InputTrack {
* with a negative timestamp should not be presented.
*/
getFirstTimestamp() {
return this._backing.getFirstTimestamp();
const result = new ResultValue<EncodedPacket | null>();
const promise = this._backing.getFirstPacket(result, { metadataOnly: true });
const getValue = () => {
const firstPacket = result.value;
return firstPacket?.timestamp ?? 0;
};
if (result.pending) {
return (promise).then(getValue);
} else {
return getValue();
}
}
/** Returns the end timestamp of the last packet of this track, in seconds. */
computeDuration() {
return this._backing.computeDuration();
computeDuration(): MaybePromise<number> {
const result = new ResultValue<EncodedPacket | null>();
const promise = this._backing.getPacket(result, Infinity, { metadataOnly: true });
const getValue = () => {
const lastPacket = result.value;
return (lastPacket?.timestamp ?? 0) + (lastPacket?.duration ?? 0);
};
if (result.pending) {
return (promise).then(getValue);
} else {
return getValue();
}
}
/**
+127 -130
View File
@@ -12,7 +12,15 @@ import { Input } from '../input';
import { InputAudioTrack, InputAudioTrackBacking } from '../input-track';
import { DEFAULT_TRACK_DISPOSITION, MetadataTags } from '../metadata';
import { PacketRetrievalOptions } from '../media-sink';
import { assert, AsyncMutex, AsyncMutex2, binarySearchExact, binarySearchLessOrEqual, ResultValue, UNDETERMINED_LANGUAGE, Yo } from '../misc';
import {
assert,
AsyncMutex4,
binarySearchExact,
binarySearchLessOrEqual,
ResultValue,
UNDETERMINED_LANGUAGE,
Yo,
} from '../misc';
import { EncodedPacket, PLACEHOLDER_DATA } from '../packet';
import { FrameHeader, getXingOffset, INFO, XING } from '../../shared/mp3-misc';
import {
@@ -42,8 +50,7 @@ export class Mp3Demuxer extends Demuxer {
tracks: InputAudioTrack[] = [];
// readingMutex = new AsyncMutex();
readingMutex = new AsyncMutex2();
readingMutex = new AsyncMutex4();
lastSampleLoaded = false;
lastLoadedPos = 0;
nextTimestampInSamples = 0;
@@ -156,66 +163,54 @@ export class Mp3Demuxer extends Demuxer {
return this.tracks;
}
async computeDuration() {
async getMetadataTags() {
using lock = this.readingMutex.lock();
if (lock.pending) await lock.ready;
await this.readMetadata();
const track = this.tracks[0];
assert(track);
return track.computeDuration();
}
async getMetadataTags() {
const release = await this.readingMutex.acquire();
try {
await this.readMetadata();
if (this.metadataTags) {
return this.metadataTags;
}
this.metadataTags = {};
let currentPos = 0;
let id3V2HeaderFound = false;
while (true) {
let headerSlice = this.reader.requestSlice(currentPos, ID3_V2_HEADER_SIZE);
if (headerSlice instanceof Promise) headerSlice = await headerSlice;
if (!headerSlice) break;
const id3V2Header = readId3V2Header(headerSlice);
if (!id3V2Header) {
break;
}
id3V2HeaderFound = true;
let contentSlice = this.reader.requestSlice(headerSlice.filePos, id3V2Header.size);
if (contentSlice instanceof Promise) contentSlice = await contentSlice;
if (!contentSlice) break;
parseId3V2Tag(contentSlice, id3V2Header, this.metadataTags);
currentPos = headerSlice.filePos + id3V2Header.size;
}
if (!id3V2HeaderFound && this.reader.fileSize !== null && this.reader.fileSize >= ID3_V1_TAG_SIZE) {
// Try reading an ID3v1 tag at the end of the file
let slice = this.reader.requestSlice(this.reader.fileSize - ID3_V1_TAG_SIZE, ID3_V1_TAG_SIZE);
if (slice instanceof Promise) slice = await slice;
assert(slice);
const tag = readAscii(slice, 3);
if (tag === 'TAG') {
parseId3V1Tag(slice, this.metadataTags);
}
}
if (this.metadataTags) {
return this.metadataTags;
} finally {
release();
}
this.metadataTags = {};
let currentPos = 0;
let id3V2HeaderFound = false;
while (true) {
let headerSlice = this.reader.requestSlice(currentPos, ID3_V2_HEADER_SIZE);
if (headerSlice instanceof Promise) headerSlice = await headerSlice;
if (!headerSlice) break;
const id3V2Header = readId3V2Header(headerSlice);
if (!id3V2Header) {
break;
}
id3V2HeaderFound = true;
let contentSlice = this.reader.requestSlice(headerSlice.filePos, id3V2Header.size);
if (contentSlice instanceof Promise) contentSlice = await contentSlice;
if (!contentSlice) break;
parseId3V2Tag(contentSlice, id3V2Header, this.metadataTags);
currentPos = headerSlice.filePos + id3V2Header.size;
}
if (!id3V2HeaderFound && this.reader.fileSize !== null && this.reader.fileSize >= ID3_V1_TAG_SIZE) {
// Try reading an ID3v1 tag at the end of the file
let slice = this.reader.requestSlice(this.reader.fileSize - ID3_V1_TAG_SIZE, ID3_V1_TAG_SIZE);
if (slice instanceof Promise) slice = await slice;
assert(slice);
const tag = readAscii(slice, 3);
if (tag === 'TAG') {
parseId3V1Tag(slice, this.metadataTags);
}
}
return this.metadataTags;
}
}
@@ -226,20 +221,11 @@ class Mp3AudioTrackBacking implements InputAudioTrackBacking {
return 1;
}
async getFirstTimestamp() {
return 0;
}
getTimeResolution() {
assert(this.demuxer.firstFrameHeader);
return this.demuxer.firstFrameHeader.sampleRate / this.demuxer.firstFrameHeader.audioSamplesInFrame;
}
async computeDuration() {
const lastPacket = await this.getPacket(Infinity, { metadataOnly: true });
return (lastPacket?.timestamp ?? 0) + (lastPacket?.duration ?? 0);
}
getName() {
return null;
}
@@ -282,7 +268,11 @@ class Mp3AudioTrackBacking implements InputAudioTrackBacking {
};
}
async getPacketAtIndex(res: ResultValue<EncodedPacket | null>, sampleIndex: number, options: PacketRetrievalOptions): Promise<Yo> {
async getPacketAtIndex(
res: ResultValue<EncodedPacket | null>,
sampleIndex: number,
options: PacketRetrievalOptions,
): Promise<Yo> {
if (sampleIndex === -1) {
return res.set(null);
}
@@ -320,80 +310,87 @@ class Mp3AudioTrackBacking implements InputAudioTrackBacking {
return this.getPacketAtIndex(res, 0, options);
}
async getNextPacket(res: ResultValue<EncodedPacket | null>, packet: EncodedPacket, options: PacketRetrievalOptions): Promise<Yo> {
// using foo = 5;
while (this.demuxer.readingMutex.locked) await this.demuxer.readingMutex.promise;
async getNextPacket(
res: ResultValue<EncodedPacket | null>,
packet: EncodedPacket,
options: PacketRetrievalOptions,
): Promise<Yo> {
using lock = this.demuxer.readingMutex.lock();
if (lock.pending) await lock.ready;
// eslint-disable-next-line @typescript-eslint/no-unused-vars
using _ = this.demuxer.readingMutex.lock();
// const release = await this.demuxer.readingMutex.acquire();
const sampleIndex = binarySearchExact(
this.demuxer.loadedSamples,
packet.timestamp,
x => x.timestamp,
);
if (sampleIndex === -1) {
throw new Error('Packet was not created from this track.');
}
try {
const sampleIndex = binarySearchExact(
const nextIndex = sampleIndex + 1;
// Ensure the next sample exists
while (
nextIndex >= this.demuxer.loadedSamples.length
&& !this.demuxer.lastSampleLoaded
) {
const result = new ResultValue<void>();
const promise = this.demuxer.advanceReader(result);
if (result.pending) await promise;
}
return this.getPacketAtIndex(res, nextIndex, options);
}
async getPacket(
res: ResultValue<EncodedPacket | null>,
timestamp: number,
options: PacketRetrievalOptions,
): Promise<Yo> {
using lock = this.demuxer.readingMutex.lock();
if (lock.pending) await lock.ready;
while (true) {
const index = binarySearchLessOrEqual(
this.demuxer.loadedSamples,
packet.timestamp,
timestamp,
x => x.timestamp,
);
if (sampleIndex === -1) {
throw new Error('Packet was not created from this track.');
if (index === -1 && this.demuxer.loadedSamples.length > 0) {
// We're before the first sample
return res.set(null);
}
const nextIndex = sampleIndex + 1;
// Ensure the next sample exists
while (
nextIndex >= this.demuxer.loadedSamples.length
&& !this.demuxer.lastSampleLoaded
) {
const result = new ResultValue<void>();
const promise = this.demuxer.advanceReader(result);
if (result.pending) await promise;
if (this.demuxer.lastSampleLoaded) {
// All data is loaded, return what we found
return this.getPacketAtIndex(res, index, options);
}
return this.getPacketAtIndex(res, nextIndex, options);
} finally {
// release();
if (index >= 0 && index + 1 < this.demuxer.loadedSamples.length) {
// The next packet also exists, we're done
return this.getPacketAtIndex(res, index, options);
}
// Otherwise, keep loading data
const result = new ResultValue<void>();
const promise = this.demuxer.advanceReader(result);
if (result.pending) await promise;
}
}
async getPacket(timestamp: number, options: PacketRetrievalOptions) {
const release = await this.demuxer.readingMutex.acquire();
try {
while (true) {
const index = binarySearchLessOrEqual(
this.demuxer.loadedSamples,
timestamp,
x => x.timestamp,
);
if (index === -1 && this.demuxer.loadedSamples.length > 0) {
// We're before the first sample
return null;
}
if (this.demuxer.lastSampleLoaded) {
// All data is loaded, return what we found
return this.getPacketAtIndex(index, options);
}
if (index >= 0 && index + 1 < this.demuxer.loadedSamples.length) {
// The next packet also exists, we're done
return this.getPacketAtIndex(index, options);
}
// Otherwise, keep loading data
await this.demuxer.advanceReader();
}
} finally {
release();
}
getKeyPacket(
res: ResultValue<EncodedPacket | null>,
timestamp: number,
options: PacketRetrievalOptions,
): Promise<Yo> {
return this.getPacket(res, timestamp, options);
}
getKeyPacket(timestamp: number, options: PacketRetrievalOptions) {
return this.getPacket(timestamp, options);
}
getNextKeyPacket(packet: EncodedPacket, options: PacketRetrievalOptions) {
return this.getNextPacket(packet, options);
getNextKeyPacket(
res: ResultValue<EncodedPacket | null>,
packet: EncodedPacket,
options: PacketRetrievalOptions,
): Promise<Yo> {
return this.getNextPacket(res, packet, options);
}
}
+22
View File
@@ -0,0 +1,22 @@
import { expect, test } from 'vitest';
import { Input } from '../../src/input.js';
import { UrlSource } from '../../src/source.js';
import { ALL_FORMATS } from '../../src/input-format.js';
import { assert } from '../../src/misc.js';
import { AudioSampleCursor } from '../../src/cursors.js';
test('MP3 decoding', async () => {
using input = new Input({
source: new UrlSource('/AudacityTest1.mp3'),
formats: ALL_FORMATS,
});
const audioTrack = await input.getPrimaryAudioTrack();
assert(audioTrack);
const cursor = new AudioSampleCursor(audioTrack);
const sample = await cursor.seekToFirst();
expect(sample).not.toBe(null);
expect(sample!.timestamp).toBe(0);
expect(sample!.duration).toBeCloseTo(1152 / audioTrack.sampleRate);
});
+95
View File
@@ -0,0 +1,95 @@
import { expect, test } from 'vitest';
import { Input } from '../../src/input.js';
import { BufferSource, FilePathSource } from '../../src/source.js';
import path from 'node:path';
import fs from 'node:fs';
import { ALL_FORMATS, MP3 } from '../../src/input-format.js';
import { PacketReader } from '../../src/cursors.js';
import { InputAudioTrack } from '../../src/input-track.js';
import { assert } from '../../src/misc.js';
import { EncodedPacket } from '../../src/packet.js';
const __dirname = new URL('.', import.meta.url).pathname;
test('MP3 demuxing', async () => {
using input = new Input({
source: new FilePathSource(path.join(__dirname, '../public/AudacityTest1.mp3')),
formats: ALL_FORMATS,
});
expect(await input.getFormat()).toBe(MP3);
expect(await input.getMimeType()).toBe('audio/mpeg');
const tracks = await input.getTracks();
expect(tracks).toHaveLength(1);
const audioTrack = tracks[0] as InputAudioTrack;
expect(audioTrack).toBeInstanceOf(InputAudioTrack);
expect(audioTrack.numberOfChannels).toBe(1);
expect(audioTrack.sampleRate).toBe(44100);
const decoderConfig = await audioTrack.getDecoderConfig();
expect(decoderConfig!.codec).toBe('mp3');
expect(decoderConfig!.numberOfChannels).toBe(audioTrack.numberOfChannels);
expect(decoderConfig!.sampleRate).toBe(audioTrack.sampleRate);
expect(decoderConfig!.description).toBeUndefined();
const reader = new PacketReader(audioTrack);
const first = await reader.readFirst();
expect(first).not.toBe(null);
expect(first!.timestamp).toBe(0);
expect(first!.data[0]).toBe(255);
const next = await reader.readNext(first!);
expect(next).not.toBe(null);
expect(next!.timestamp).toBeCloseTo(first!.timestamp + first!.duration);
expect(next!.sequenceNumber).toBeGreaterThan(first!.sequenceNumber);
const nextKey = await reader.readNextKey(first!);
expect(nextKey).not.toBe(null);
expect(nextKey!.timestamp).toBe(next!.timestamp);
const seeked = await reader.readAt(1);
expect(seeked).not.toBe(null);
expect(seeked!.timestamp).toBeGreaterThan(0.9);
expect(seeked!.timestamp).toBeLessThanOrEqual(1);
expect(seeked!.sequenceNumber).toBeGreaterThan(next!.sequenceNumber);
const seekedKey = await reader.readKeyAt(1);
expect(seekedKey).not.toBe(null);
expect(seeked!.sequenceNumber).toBe(seeked!.sequenceNumber);
const last = await reader.readAt(Infinity);
expect(last).not.toBe(null);
expect(last!.sequenceNumber).toBeGreaterThan(seeked!.sequenceNumber);
const afterLast = await reader.readNext(last!);
expect(afterLast).toBe(null);
const firstTimestamp = await audioTrack.getFirstTimestamp();
expect(firstTimestamp).toBe(0);
const duration = await audioTrack.computeDuration();
expect(duration).toBe(last!.timestamp + last!.duration);
});
test('Sync reading', async () => {
using input = new Input({
source: new BufferSource(fs.readFileSync(path.join(__dirname, '../public/AudacityTest1.mp3'))),
formats: ALL_FORMATS,
});
const audioTrack = await input.getPrimaryAudioTrack();
assert(audioTrack);
const reader = new PacketReader(audioTrack);
let current = reader.readFirst() as EncodedPacket | null;
let count = 0;
while (current) {
current = reader.readNext(current) as EncodedPacket | null;
count++;
}
expect(count).toBe(475);
});
Binary file not shown.