diff --git a/dev/demux.html b/dev/demux.html index 8d02ae1..788a0cf 100644 --- a/dev/demux.html +++ b/dev/demux.html @@ -14,7 +14,61 @@ source: new Mediabunny.BlobSource(file), }); - console.log(await input.getTracks()); + const videoTrack = await input.getPrimaryVideoTrack(); + const cursor = await Mediabunny.VideoSampleCursor2.init(videoTrack); + + /* + console.log(await cursor.seekTo(2.00)); + console.log(await cursor.next()); + console.log(await cursor.next()); + */ + + const mh = [ + cursor.seekTo(2.10), + cursor.seekTo(5), + //cursor.seekTo(2.00), + ]; + + console.log(await Promise.all(mh)); + + //console.log(await cursor.seekTo(2.05)); + //console.log(await cursor.seekTo(2.10)); + //console.log(await cursor.seekTo(10.2)); + //console.log(await cursor.seekTo(5.2)); + + /* + setTimeout(async () => { + console.log("this") + + }, 500); + */ + + + /* + const iters = 5; + let total = 0; + + for (let i = 0; i < iters; i++) { + const file = fileInput.files[0]; + const input = new Mediabunny.Input({ + formats: Mediabunny.ALL_FORMATS, + source: new Mediabunny.BufferSource(await file.arrayBuffer()), + }); + + const audioTrack = await input.getPrimaryAudioTrack(); + const cursor = new Mediabunny.PacketCursor(audioTrack); + + await cursor.seekToStart(); + + const start = performance.now(); + await cursor.iterate(() => {}); + //alert(performance.now() - start) + + total += performance.now() - start; + } + + alert(total / iters); + */ /* return; @@ -95,6 +149,8 @@ total += end - start; //console.log(total / file.size, end - start); } + alert(elapsed / iters); + const videoTrack = await input.getPrimaryVideoTrack(); const sink = new Mediabunny.EncodedPacketSink(videoTrack); diff --git a/scripts/bundle.ts b/scripts/bundle.ts index fbe7a91..cf8d7e0 100644 --- a/scripts/bundle.ts +++ b/scripts/bundle.ts @@ -16,6 +16,7 @@ const createVariants = async ( entryPoints: [entryPoint], bundle: true, logLevel: 'info', + target: 'es2024', // Keep? This makes sure "using" is transformed logOverride: { 'import-is-undefined': 'silent', // Warning caused by the disabled "node.ts" import }, diff --git a/src/cursors.ts b/src/cursors.ts new file mode 100644 index 0000000..58dd446 --- /dev/null +++ b/src/cursors.ts @@ -0,0 +1,612 @@ +// Two parallelism modes: Cancel and queue (right?) +// Useful in packet context? Or only for sample? + +import { InputTrack, InputVideoTrack } from './input-track'; +import { PacketRetrievalOptions, VideoDecoderWrapper } from './media-sink'; +import { assert, assertNever, AsyncMutex, AsyncMutex2, insertSorted, last, MaybePromise, promiseWithResolvers, ResultValue, Yo } from './misc'; +import { EncodedPacket } from './packet'; +import { VideoSample } from './sample'; + +export class PacketCursor { + track: InputTrack; + _options: PacketRetrievalOptions; + current: EncodedPacket | null = null; + initialized = false; + + constructor(track: InputTrack, options: PacketRetrievalOptions = {}) { + this.track = track; + this._options = options; + } + + peekAtStart(): MaybePromise { + const result = new ResultValue(); + const promise = this.track._backing.getFirstPacket(result, this._options); + + if (result.pending) { + return (promise as Promise).then(() => result.value); + } else { + return result.value; + } + } + + seekToStart(): MaybePromise { + const result = new ResultValue(); + const promise = this.track._backing.getFirstPacket(result, this._options); + + if (result.pending) { + return (promise as Promise).then(() => { + this.initialized = true; + return this.current = result.value; + }); + } else { + this.initialized = true; + return this.current = result.value; + } + } + + peekAt(timestamp: number): MaybePromise { + const result = new ResultValue(); + const promise = this.track._backing.getPacket(result, timestamp, this._options); + + if (result.pending) { + return (promise as Promise).then(() => result.value); + } else { + return result.value; + } + } + + seekTo(timestamp: number): MaybePromise { + const result = new ResultValue(); + const promise = this.track._backing.getPacket(result, timestamp, this._options); + + if (result.pending) { + return (promise as Promise).then(() => { + this.initialized = true; + return this.current = result.value; + }); + } else { + this.initialized = true; + return this.current = result.value; + } + } + + peekKeyAt(timestamp: number): MaybePromise { + const result = new ResultValue(); + const promise = this.track._backing.getKeyPacket(result, timestamp, this._options); + + if (result.pending) { + return (promise as Promise).then(() => result.value); + } else { + return result.value; + } + } + + seekToKey(timestamp: number): MaybePromise { + const result = new ResultValue(); + const promise = this.track._backing.getKeyPacket(result, timestamp, this._options); + + if (result.pending) { + return (promise as Promise).then(() => { + this.initialized = true; + return this.current = result.value; + }); + } else { + this.initialized = true; + return this.current = result.value; + } + } + + _ensureInitialized() { + if (!this.initialized) { + throw new Error('You must first initialize the cursor to a position by calling any of the seek methods.'); + } + } + + next(): MaybePromise { + this._ensureInitialized(); + + if (!this.current) { + return null; + } + + const result = new ResultValue(); + const promise = this.track._backing.getNextPacket(result, this.current, this._options); + + if (result.pending) { + return (promise as Promise).then(() => { + return this.current = result.value; + }); + } else { + return this.current = result.value; + } + } + + peekNextKey(): MaybePromise { + this._ensureInitialized(); + + if (!this.current) { + return null; + } + + const result = new ResultValue(); + const promise = this.track._backing.getNextKeyPacket(result, this.current, this._options); + + if (result.pending) { + return (promise as Promise).then(() => result.value); + } else { + return result.value; + } + } + + nextKey(): MaybePromise { + this._ensureInitialized(); + + if (!this.current) { + return null; + } + + const result = new ResultValue(); + const promise = this.track._backing.getNextKeyPacket(result, this.current, this._options); + + if (result.pending) { + return (promise as Promise).then(() => { + return this.current = result.value; + }); + } else { + return this.current = result.value; + } + } + + async iterate(callback: (packet: EncodedPacket, stop: () => void) => MaybePromise) { + this._ensureInitialized(); + + let stopped = false; + const stop = () => stopped = true; + + while (this.current) { + const result = callback(this.current, stop); + if (result instanceof Promise) await result; + + if (stopped) { + break; + } + + let next = this.next(); + if (next instanceof Promise) next = await next; + + this.current = next; + } + } + + // eslint-disable-next-line @stylistic/generator-star-spacing + async *[Symbol.asyncIterator]() { + this._ensureInitialized(); + + while (this.current) { + yield this.current; + + let next = this.next(); + if (next instanceof Promise) next = await next; + + this.current = next; + } + } +} + +export class VideoSampleCursor2 { + track: InputVideoTrack; + initialized = false; + packetCursor: PacketCursor; + pumpRunning = false; + decoder: VideoDecoderWrapper; + // current: VideoSample | null = null; + sampleQueue: VideoSample[] = []; + queueDequeue = promiseWithResolvers(); + pendingRequests: { + timestamp: number; + resolve: (sample: VideoSample | null) => void; + }[] = []; + + stopPump = false; + pumpStopped = promiseWithResolvers(); + + decodedTimestamps: number[] = []; + maxDecodedSequenceNumber = -1; + pumpMutex = new AsyncMutex2(); + + private constructor(track: InputVideoTrack, decoder: VideoDecoderWrapper) { + this.track = track; + this.decoder = decoder; + this.packetCursor = new PacketCursor(track); + } + + static async init(track: InputVideoTrack) { + if (!(await track.canDecode())) { + throw new Error( + 'This video track cannot be decoded by this browser. Make sure to check decodability before using' + + ' a track.', + ); + } + + const decoderConfig = await track.getDecoderConfig(); + assert(decoderConfig); + assert(track.codec); + + const decoder = new VideoDecoderWrapper( + (sample) => { + while (cursor.decodedTimestamps.length > 0 && cursor.decodedTimestamps[0]! <= sample.timestamp) { + cursor.decodedTimestamps.shift(); + } + + if (cursor.stopPump) { + sample.close(); + return; + } + + console.log('revc', sample.timestamp); + + if (cursor.pendingRequests.length === 0) { + cursor.sampleQueue.push(sample); + } else { + for (let i = 0; i < cursor.pendingRequests.length; i++) { + const request = cursor.pendingRequests[i]!; + if (request.timestamp <= sample.timestamp) { + request.resolve(sample.clone()); + cursor.pendingRequests.splice(i--, 1); + } + } + } + + // cursor.current?.close(); + // cursor.current = sample; + + cursor.queueDequeue.resolve(); + cursor.queueDequeue = promiseWithResolvers(); + }, + (error) => { + console.error(error); + }, + track.codec, + decoderConfig, + track.rotation, + track.timeResolution, + ); + + const cursor = new VideoSampleCursor2(track, decoder); + return cursor; + } + + getNextExpectedTimestamp() { + if (this.sampleQueue.length > 0) { + return this.sampleQueue[0]!.timestamp; + } + } + + async seekTo(timestamp: number): Promise { + this.initialized = true; // too late? + + console.log('a'); + while (this.pumpMutex.locked) { + console.log('waiting...'); + await this.pumpMutex.promise; + } + + console.log('GOIN IN'); + + using _ = this.pumpMutex.lock(); + + const targetPacket = await this.packetCursor.peekAt(timestamp); + if (!targetPacket) { + return null; + } + + let setNewPump = true; + + if (this.sampleQueue.length > 0 && targetPacket.timestamp <= this.sampleQueue[0]!.timestamp) { + console.log('This bitch case kicked'); + } else { + while (this.sampleQueue.length > 0) { + const nextSample = this.sampleQueue[0]!; + if (targetPacket.timestamp <= nextSample.timestamp) { + console.log('used this path'); + return nextSample; + } + + this.sampleQueue.shift(); + this.queueDequeue.resolve(); + this.queueDequeue = promiseWithResolvers(); + } + + if (this.maxDecodedSequenceNumber !== -1) { + // This means a packet was queued for decode and the cursor is initialized + + if (targetPacket.sequenceNumber <= this.maxDecodedSequenceNumber) { + const nextExpectedTimestamp = this.decodedTimestamps[0]; + if (!nextExpectedTimestamp || nextExpectedTimestamp > timestamp) { + // yeah + } else { + setNewPump = false; + } + } else { + const key = await this.packetCursor.peekNextKey(); + if (!key || targetPacket.sequenceNumber < key.sequenceNumber) { + setNewPump = false; + } + } + } + } + + if (setNewPump) { + console.log('setting up a new PUMP'); + + if (this.pumpRunning) { + this.stopPump = true; + this.queueDequeue.resolve(); + this.queueDequeue = promiseWithResolvers(); + await this.pumpStopped.promise; + + for (const sample of this.sampleQueue) { + sample.close(); + } + this.sampleQueue.length = 0; + this.maxDecodedSequenceNumber = -1; + this.decodedTimestamps.length = 0; + this.stopPump = false; + } + + await this.packetCursor.seekToKey(timestamp); + void this.runPump(); + await Promise.resolve(); // lol + } + + const request = promiseWithResolvers(); + this.pendingRequests.push({ + timestamp: targetPacket.timestamp, + resolve: request.resolve, + }); + + return request.promise; + } + + async next() { + while (this.pumpMutex.locked) { + console.log('waiting next...'); + await this.pumpMutex.promise; + } + + if (!this.initialized) { + throw new Error('This shud be the indicator the next not being available I think'); + } + + if (this.sampleQueue.length > 0) { + const nextSample = this.sampleQueue.shift()!; + this.queueDequeue.resolve(); + this.queueDequeue = promiseWithResolvers(); + + return nextSample; + } + + if (!this.pumpRunning) { + return null; // None more after this, boy + } + + const request = promiseWithResolvers(); + this.pendingRequests.push({ + timestamp: -Infinity, // Matches any sample timestamp, so any next one will match + resolve: request.resolve, + }); + + return request.promise; + } + + async runPump() { + assert(this.packetCursor.current); + + this.pumpRunning = true; + + while (this.packetCursor.current && !this.stopPump) { + const maxQueueSize = 8 ?? computeMaxQueueSize(this.sampleQueue.length); // temp + if (this.sampleQueue.length + this.decoder.getDecodeQueueSize() > maxQueueSize) { + await this.queueDequeue.promise; + continue; + } + + insertSorted(this.decodedTimestamps, this.packetCursor.current.timestamp, x => x); + this.maxDecodedSequenceNumber = this.packetCursor.current.sequenceNumber; + this.decoder.decode(this.packetCursor.current); + await this.packetCursor.next(); + } + + console.log('stopping current pump...'); + await this.decoder.flush(); + + this.pumpStopped.resolve(); + this.pumpStopped = promiseWithResolvers(); + + this.pumpRunning = false; + + this.pendingRequests.forEach(x => x.resolve(null)); + this.pendingRequests.length = 0; + } +} + +async function weJustTesting() { + const cursor = new VideoSampleCursor2(); + + // Spins up decoder and resolves to the sample + await cursor.seekTo(2); + + // This can do multiple things: + // - It pops its internal sample queue until it finds a matching frame; in this case, it returns instantly (no promise) + // - If that wasn't possible, but the packet that corresponds to the requested sample was already queued for encoding, + // it will wait for the decoder to spit it out and then returns it. I guess this requires a "pending requests" ahh + // structure somewhere. + // - If that's also not the case, but the seeked packet is in the current GOP, then it just keeps pumping packets into + // the decoder. + // - If the requested packet is outside of the current GOP or "backwards" from the current stream, it resets the + // internal decoder + // In any case, there's always a "pump" running that supplies the decoder with new packets to decode. This pump is + // halted if the internal queue is sufficiently large, and is resumed when samples are consumed. + // This pump is reset if necessary. + await cursor.seekTo(2.1); +} + +export class VideoSampleCursor { + track: InputVideoTrack; + current: EncodedPacket | null = null; + initialized = false; + packetCursor: PacketCursor; + packetCursor2: PacketCursor; + + decoder!: VideoDecoderWrapper; + + constructor(track: InputVideoTrack) { + this.track = track; + this.packetCursor = new PacketCursor(track, { verifyKeyPackets: true }); + this.packetCursor2 = new PacketCursor(track, { verifyKeyPackets: true }); // not good + } + + async init() { + if (!(await this.track.canDecode())) { + throw new Error( + 'This video track cannot be decoded by this browser. Make sure to check decodability before using' + + ' a track.', + ); + } + + const decoderConfig = await this.track.getDecoderConfig(); + + this.decoder = new VideoDecoderWrapper( + (sample) => { + this.sampleQueue.push(sample); + }, + (error) => { + // Un que? + }, + this.track.codec!, + decoderConfig!, + this.track.rotation, + this.track.timeResolution, + ); + } + + pumpFinished = promiseWithResolvers(); + queueDequeue = promiseWithResolvers(); + terminatePump = false; + pumpRunning = false; + sampleQueue: VideoSample[] = []; + + async runPump() { + this.pumpRunning = true; + + while (this.packetCursor.current && !this.terminatePump) { + const maxQueueSize = computeMaxQueueSize(0); + if (0 + this.decoder.getDecodeQueueSize() > maxQueueSize) { + this.queueDequeue = promiseWithResolvers(); + await this.queueDequeue.promise; + continue; + } + + this.decoder.decode(this.packetCursor.current); + + const result = this.packetCursor.next(); + if (result instanceof Promise) await result; + } + + await this.decoder.flush(); + this.pumpFinished.resolve(); + this.pumpRunning = false; + } + + async beginNewRun() { + if (this.pumpRunning) { + this.terminatePump = true; + await this.pumpFinished.promise; + + for (const sample of this.sampleQueue) { + sample.close(); + } + this.sampleQueue.length = 0; + } + + // todo errors + void this.runPump(); + } + + async _seekToCurrentPacket(res: ResultValue): Promise { + const targetPacket = this.packetCursor.current; + assert(targetPacket); + + if (targetPacket.type !== 'key') { + await this.packetCursor.seekToKey(targetPacket.timestamp); + } + } + + /* + seekToStart(): MaybePromise { + const onPacket = (packet: EncodedPacket | null) => { + if (!packet) { + return null; + } + + const result = new ResultValue(); + const promise = this._seekToPacket(result, packet); + + if (result.pending) { + return promise.then(() => result.value); + } else { + return result.value; + } + }; + + const packet = this.packetCursor.seekToStart(); + if (packet instanceof Promise) { + return packet.then(onPacket); + } else { + return onPacket(packet); + } + } + */ + + async seekTo(timestamp: number): Promise { + const packet = await this.packetCursor2.seekTo(timestamp); + if (!packet) { + return null; // I guess? + } + + if (this.packetCursor.current) { + // if (packet.sequenceNumber) + } + + /* + const onPacket = (packet: EncodedPacket | null) => { + if (!packet) { + return null; + } + + const result = new ResultValue(); + const promise = this._seekToPacket(result, packet); + + if (result.pending) { + return promise.then(() => result.value); + } else { + return result.value; + } + }; + + const packet = this.packetCursor.seekTo(timestamp); + if (packet instanceof Promise) { + return packet.then(onPacket); + } else { + return onPacket(packet); + } + */ + } +} + +const computeMaxQueueSize = (decodedSampleQueueSize: number) => { + // If we have decoded samples lying around, limit the total queue size to a small value (decoded samples can use up + // a lot of memory). If not, we're fine with a much bigger queue of encoded packets waiting to be decoded. In fact, + // some decoders only start flushing out decoded chunks when the packet queue is large enough. + return decodedSampleQueueSize === 0 ? 40 : 8; +}; diff --git a/src/index.ts b/src/index.ts index f5dbdf1..ec6454a 100644 --- a/src/index.ts +++ b/src/index.ts @@ -182,6 +182,10 @@ export { WrappedAudioBuffer, WrappedCanvas, } from './media-sink'; +export { + PacketCursor, + VideoSampleCursor2, +} from './cursors'; export { Conversion, ConversionOptions, diff --git a/src/input-format.ts b/src/input-format.ts index 1748f27..7a20c89 100644 --- a/src/input-format.ts +++ b/src/input-format.ts @@ -259,6 +259,7 @@ export class WebMInputFormat extends MatroskaInputFormat { export class Mp3InputFormat extends InputFormat { /** @internal */ async _canReadInput(input: Input) { + return true; let slice = input._reader.requestSlice(0, 10); if (slice instanceof Promise) slice = await slice; if (!slice) return false; diff --git a/src/input-track.ts b/src/input-track.ts index 8c137fc..daac518 100644 --- a/src/input-track.ts +++ b/src/input-track.ts @@ -11,7 +11,7 @@ import { determineVideoPacketType } from './codec-data'; import { customAudioDecoders, customVideoDecoders } from './custom-coder'; import { Input } from './input'; import { EncodedPacketSink, PacketRetrievalOptions } from './media-sink'; -import { assert, Rotation } from './misc'; +import { assert, MaybePromise, ResultValue, Rotation, Yo } from './misc'; import { TrackType } from './output'; import { EncodedPacket, PacketType } from './packet'; import { TrackDisposition } from './metadata'; @@ -41,11 +41,17 @@ export interface InputTrackBacking { getFirstTimestamp(): Promise; computeDuration(): Promise; - getFirstPacket(options: PacketRetrievalOptions): Promise; - getPacket(timestamp: number, options: PacketRetrievalOptions): Promise; - getNextPacket(packet: EncodedPacket, options: PacketRetrievalOptions): Promise; - getKeyPacket(timestamp: number, options: PacketRetrievalOptions): Promise; - getNextKeyPacket(packet: EncodedPacket, options: PacketRetrievalOptions): Promise; + getFirstPacket(res: ResultValue, options: PacketRetrievalOptions): MaybePromise; + getNextPacket(res: ResultValue, packet: EncodedPacket, options: PacketRetrievalOptions): MaybePromise; + getPacket(res: ResultValue, timestamp: number, options: PacketRetrievalOptions): MaybePromise; + getKeyPacket(res: ResultValue, timestamp: number, options: PacketRetrievalOptions): MaybePromise; + getNextKeyPacket(res: ResultValue, packet: EncodedPacket, options: PacketRetrievalOptions): MaybePromise; + + // getFirstPacket(options: PacketRetrievalOptions): Promise; + // getPacket(timestamp: number, options: PacketRetrievalOptions): Promise; + // getNextPacket(packet: EncodedPacket, options: PacketRetrievalOptions): Promise; + // getKeyPacket(timestamp: number, options: PacketRetrievalOptions): Promise; + // getNextKeyPacket(packet: EncodedPacket, options: PacketRetrievalOptions): Promise; } /** diff --git a/src/isobmff/isobmff-demuxer.ts b/src/isobmff/isobmff-demuxer.ts index b76fd85..be4e1db 100644 --- a/src/isobmff/isobmff-demuxer.ts +++ b/src/isobmff/isobmff-demuxer.ts @@ -58,6 +58,9 @@ import { UNDETERMINED_LANGUAGE, toDataView, roundIfAlmostInteger, + ResultValue, + Yo, + MaybePromise, } from '../misc'; import { EncodedPacket, PLACEHOLDER_DATA } from '../packet'; import { buildIsobmffMimeType } from './isobmff-misc'; @@ -2366,13 +2369,22 @@ abstract class IsobmffTrackBacking implements InputTrackBacking { return firstPacket?.timestamp ?? 0; } - async getFirstPacket(options: PacketRetrievalOptions) { - const regularPacket = await this.fetchPacketForSampleIndex(0, options); - if (regularPacket || !this.internalTrack.demuxer.isFragmented) { + async getFirstPacket(res: ResultValue, options: PacketRetrievalOptions): Promise { + const promise = this.fetchPacketForSampleIndex(res, 0, options); + if (res.pending) await promise; + + // const regularPacket = res.value; + + // const regularPacket = await this.fetchPacketForSampleIndex(0, options); + if (res.value || !this.internalTrack.demuxer.isFragmented) { // If there's a non-fragmented packet, always prefer that - return regularPacket; + return res.pass(); + // return regularPacket; } + throw new Error('bruh'); + + /* return this.performFragmentedLookup( null, (fragment) => { @@ -2393,6 +2405,7 @@ abstract class IsobmffTrackBacking implements InputTrackBacking { Infinity, options, ); + */ } private mapTimestampIntoTimescale(timestamp: number) { @@ -2402,18 +2415,27 @@ abstract class IsobmffTrackBacking implements InputTrackBacking { return roundIfAlmostInteger(timestamp * this.internalTrack.timescale) + this.internalTrack.editListOffset; } - async getPacket(timestamp: number, options: PacketRetrievalOptions) { + async getPacket( + res: ResultValue, + timestamp: number, + options: PacketRetrievalOptions, + ): Promise { const timestampInTimescale = this.mapTimestampIntoTimescale(timestamp); const sampleTable = this.internalTrack.demuxer.getSampleTableForTrack(this.internalTrack); const sampleIndex = getSampleIndexForTimestamp(sampleTable, timestampInTimescale); - const regularPacket = await this.fetchPacketForSampleIndex(sampleIndex, options); + + const promise = this.fetchPacketForSampleIndex(res, sampleIndex, options); + if (res.pending) await promise; if (!sampleTableIsEmpty(sampleTable) || !this.internalTrack.demuxer.isFragmented) { // Prefer the non-fragmented packet - return regularPacket; + return res.pass(); } + throw new Error('how'); + + /* return this.performFragmentedLookup( null, (fragment) => { @@ -2437,16 +2459,24 @@ abstract class IsobmffTrackBacking implements InputTrackBacking { timestampInTimescale, options, ); + */ } - async getNextPacket(packet: EncodedPacket, options: PacketRetrievalOptions) { - const regularSampleIndex = this.packetToSampleIndex.get(packet); + async getNextPacket( + res: ResultValue, + packet: EncodedPacket, + options: PacketRetrievalOptions, + ): Promise { + const regularSampleIndex = packet.sampleIndex;// this.packetToSampleIndex.get(packet); if (regularSampleIndex !== undefined) { // Prefer the non-fragmented packet - return this.fetchPacketForSampleIndex(regularSampleIndex + 1, options); + return this.fetchPacketForSampleIndex(res, regularSampleIndex + 1, options); } + throw new Error('This'); + + /* const locationInFragment = this.packetToFragmentLocation.get(packet); if (locationInFragment === undefined) { throw new Error('Packet was not created from this track.'); @@ -2483,9 +2513,14 @@ abstract class IsobmffTrackBacking implements InputTrackBacking { Infinity, options, ); + */ } - async getKeyPacket(timestamp: number, options: PacketRetrievalOptions) { + async getKeyPacket( + res: ResultValue, + timestamp: number, + options: PacketRetrievalOptions, + ): Promise { const timestampInTimescale = this.mapTimestampIntoTimescale(timestamp); const sampleTable = this.internalTrack.demuxer.getSampleTableForTrack(this.internalTrack); @@ -2493,13 +2528,19 @@ abstract class IsobmffTrackBacking implements InputTrackBacking { const keyFrameSampleIndex = sampleIndex === -1 ? -1 : getRelevantKeyframeIndexForSample(sampleTable, sampleIndex); - const regularPacket = await this.fetchPacketForSampleIndex(keyFrameSampleIndex, options); + + const promise = this.fetchPacketForSampleIndex(res, keyFrameSampleIndex, options); + if (res.pending) await promise; if (!sampleTableIsEmpty(sampleTable) || !this.internalTrack.demuxer.isFragmented) { // Prefer the non-fragmented packet - return regularPacket; + return res.pass(); } + throw new Error('minusch'); + + /* + return this.performFragmentedLookup( null, (fragment) => { @@ -2522,15 +2563,20 @@ abstract class IsobmffTrackBacking implements InputTrackBacking { timestampInTimescale, options, ); + */ } - async getNextKeyPacket(packet: EncodedPacket, options: PacketRetrievalOptions) { - const regularSampleIndex = this.packetToSampleIndex.get(packet); + async getNextKeyPacket( + res: ResultValue, + packet: EncodedPacket, + options: PacketRetrievalOptions, + ): Promise { + const regularSampleIndex = packet.sampleIndex;// this.packetToSampleIndex.get(packet); if (regularSampleIndex !== undefined) { // Prefer the non-fragmented packet const sampleTable = this.internalTrack.demuxer.getSampleTableForTrack(this.internalTrack); const nextKeyFrameSampleIndex = getNextKeyframeIndexForSample(sampleTable, regularSampleIndex); - return this.fetchPacketForSampleIndex(nextKeyFrameSampleIndex, options); + return this.fetchPacketForSampleIndex(res, nextKeyFrameSampleIndex, options); } const locationInFragment = this.packetToFragmentLocation.get(packet); @@ -2538,6 +2584,9 @@ abstract class IsobmffTrackBacking implements InputTrackBacking { throw new Error('Packet was not created from this track.'); } + throw new Error('Sie sagen Mel'); + + /* return this.performFragmentedLookup( locationInFragment.fragment, (fragment) => { @@ -2576,17 +2625,22 @@ abstract class IsobmffTrackBacking implements InputTrackBacking { Infinity, options, ); + */ } - private async fetchPacketForSampleIndex(sampleIndex: number, options: PacketRetrievalOptions) { + private async fetchPacketForSampleIndex( + res: ResultValue, + sampleIndex: number, + options: PacketRetrievalOptions, + ): Promise { if (sampleIndex === -1) { - return null; + return res.set(null); } const sampleTable = this.internalTrack.demuxer.getSampleTableForTrack(this.internalTrack); const sampleInfo = getSampleInfo(sampleTable, sampleIndex); if (!sampleInfo) { - return null; + return res.set(null); } let data: Uint8Array; @@ -2615,9 +2669,10 @@ abstract class IsobmffTrackBacking implements InputTrackBacking { sampleInfo.sampleSize, ); - this.packetToSampleIndex.set(packet, sampleIndex); + packet.sampleIndex = sampleIndex; + // this.packetToSampleIndex.set(packet, sampleIndex); - return packet; + return res.set(packet); } private async fetchPacketInFragment(fragment: Fragment, sampleIndex: number, options: PacketRetrievalOptions) { @@ -2922,8 +2977,6 @@ type SampleInfo = { duration: number; sampleOffset: number; sampleSize: number; - chunkOffset: number; - chunkSize: number; isKeyFrame: boolean; }; @@ -2958,20 +3011,13 @@ const getSampleInfo = (sampleTable: SampleTable, sampleIndex: number): SampleInf const startSampleIndexOfChunk = chunkEntry.startSampleIndex + (chunkIndex - chunkEntry.startChunkIndex) * chunkEntry.samplesPerChunk; - let chunkSize = 0; let sampleOffset = chunkOffset; if (sampleTable.sampleSizes.length === 1) { sampleOffset += sampleSize * (sampleIndex - startSampleIndexOfChunk); - chunkSize += sampleSize * chunkEntry.samplesPerChunk; } else { - for (let i = startSampleIndexOfChunk; i < startSampleIndexOfChunk + chunkEntry.samplesPerChunk; i++) { - const sampleSize = sampleTable.sampleSizes[i]!; - - if (i < sampleIndex) { - sampleOffset += sampleSize; - } - chunkSize += sampleSize; + for (let i = startSampleIndexOfChunk; i < sampleIndex; i++) { + sampleOffset += sampleTable.sampleSizes[i]!; } } @@ -2994,8 +3040,6 @@ const getSampleInfo = (sampleTable: SampleTable, sampleIndex: number): SampleInf duration, sampleOffset, sampleSize, - chunkOffset, - chunkSize, isKeyFrame: sampleTable.keySampleIndices ? binarySearchExact(sampleTable.keySampleIndices, sampleIndex, x => x) !== -1 : true, diff --git a/src/media-sink.ts b/src/media-sink.ts index 841d8f5..db36e60 100644 --- a/src/media-sink.ts +++ b/src/media-sink.ts @@ -32,12 +32,15 @@ import { isWebKit, last, mapAsyncGenerator, + MaybePromise, promiseWithResolvers, + ResultValue, Rotation, toAsyncIterator, toDataView, toUint8Array, validateAnyIterable, + Yo, } from './misc'; import { EncodedPacket } from './packet'; import { fromAlaw, fromUlaw } from './pcm'; @@ -131,14 +134,20 @@ export class EncodedPacketSink { * Retrieves the track's first packet (in decode order), or null if it has no packets. The first packet is very * likely to be a key packet. */ - getFirstPacket(options: PacketRetrievalOptions = {}) { + getFirstPacket(options: PacketRetrievalOptions = {}): MaybePromise { validatePacketRetrievalOptions(options); if (this._track.input._disposed) { throw new InputDisposedError(); } - return maybeFixPacketType(this._track, this._track._backing.getFirstPacket(options), options); + const result = new ResultValue(); + const promise = this._track._backing.getFirstPacket(result, options); + if (result.pending) return (promise as Promise).then(() => result.value); + + return result.value; + + // return maybeFixPacketType(this._track, this._track._backing.getFirstPacket(options), options); } /** @@ -174,7 +183,13 @@ export class EncodedPacketSink { throw new InputDisposedError(); } - return maybeFixPacketType(this._track, this._track._backing.getNextPacket(packet, options), options); + const result = new ResultValue(); + const promise = this._track._backing.getNextPacket(result, packet, options); + if (result.pending) return (promise as Promise).then(() => result.value); + + return result.value; + + // return maybeFixPacketType(this._track, this._track._backing.getNextPacket(packet, options), options); } /** @@ -809,7 +824,7 @@ const computeMaxQueueSize = (decodedSampleQueueSize: number) => { return decodedSampleQueueSize === 0 ? 40 : 8; }; -class VideoDecoderWrapper extends DecoderWrapper { +export class VideoDecoderWrapper extends DecoderWrapper { decoder: VideoDecoder | null = null; customDecoder: CustomVideoDecoder | null = null; diff --git a/src/misc.ts b/src/misc.ts index 68cb687..b893522 100644 --- a/src/misc.ts +++ b/src/misc.ts @@ -796,3 +796,46 @@ export const polyfillSymbolDispose = () => { export const isNumber = (x: unknown) => { return typeof x === 'number' && !Number.isNaN(x); }; + +// eslint-disable-next-line @typescript-eslint/no-unused-vars +const yo: unique symbol = Symbol(); +export type Yo = typeof yo; + +export class ResultValue { + value!: T; + pending = true; + + // @ts-expect-error Just for the types + set(value: T): Yo { + this.value = value; + this.pending = false; + } + + // @ts-expect-error Just for the types + pass(): Yo { + assert(!this.pending); + } +} + +export class AsyncMutex2 { + locked = false; + promise = Promise.resolve(); + + lock() { + if (this.locked) { + throw new Error('Mutex already locked.'); + } + + this.locked = true; + + const { promise, resolve } = promiseWithResolvers(); + this.promise = promise; + + return { + [Symbol.dispose]: () => { + resolve(); + this.locked = false; + }, + }; + } +} diff --git a/src/mp3/mp3-demuxer.ts b/src/mp3/mp3-demuxer.ts index fe654d8..8b78fcd 100644 --- a/src/mp3/mp3-demuxer.ts +++ b/src/mp3/mp3-demuxer.ts @@ -12,7 +12,7 @@ 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, binarySearchExact, binarySearchLessOrEqual, UNDETERMINED_LANGUAGE } from '../misc'; +import { assert, AsyncMutex, AsyncMutex2, 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,7 +42,8 @@ export class Mp3Demuxer extends Demuxer { tracks: InputAudioTrack[] = []; - readingMutex = new AsyncMutex(); + // readingMutex = new AsyncMutex(); + readingMutex = new AsyncMutex2(); lastSampleLoaded = false; lastLoadedPos = 0; nextTimestampInSamples = 0; @@ -57,7 +58,9 @@ export class Mp3Demuxer extends Demuxer { return this.metadataPromise ??= (async () => { // Keep loading until we find the first frame header while (!this.firstFrameHeader && !this.lastSampleLoaded) { - await this.advanceReader(); + const result = new ResultValue(); + const promise = this.advanceReader(result); + if (result.pending) await promise; } if (!this.firstFrameHeader) { @@ -68,7 +71,7 @@ export class Mp3Demuxer extends Demuxer { })(); } - async advanceReader() { + async advanceReader(res: ResultValue): Promise { if (this.lastLoadedPos === 0) { // Let's skip all ID3v2 tags at the start of the file while (true) { @@ -77,7 +80,7 @@ export class Mp3Demuxer extends Demuxer { if (!slice) { this.lastSampleLoaded = true; - return; + return res.set(); } const id3V2Header = readId3V2Header(slice); @@ -89,19 +92,25 @@ export class Mp3Demuxer extends Demuxer { } } - const result = await readNextFrameHeader(this.reader, this.lastLoadedPos, this.reader.fileSize); - if (!result) { + const result = new ResultValue<{ + header: FrameHeader; + startPos: number; + } | null>(); + const promise = readNextFrameHeader(result, this.reader, this.lastLoadedPos, this.reader.fileSize); + if (result.pending) await promise; + + if (!result.value) { this.lastSampleLoaded = true; - return; + return res.set(); } - const header = result.header; + const header = result.value.header; - this.lastLoadedPos = result.startPos + header.totalSize - 1; // -1 in case the frame is 1 byte too short + this.lastLoadedPos = result.value.startPos + header.totalSize - 1; // -1 in case the frame is 1 byte too short const xingOffset = getXingOffset(header.mpegVersionId, header.channel); - let slice = this.reader.requestSlice(result.startPos + xingOffset, 4); + let slice = this.reader.requestSlice(result.value.startPos + xingOffset, 4); if (slice instanceof Promise) slice = await slice; if (slice) { const word = readU32Be(slice); @@ -109,7 +118,7 @@ export class Mp3Demuxer extends Demuxer { if (isXing) { // There's no actual audio data in this frame, so let's skip it - return; + return res.set(); } } @@ -128,14 +137,14 @@ export class Mp3Demuxer extends Demuxer { const sample: Sample = { timestamp: this.nextTimestampInSamples / this.firstFrameHeader.sampleRate, duration: sampleDuration, - dataStart: result.startPos, + dataStart: result.value.startPos, dataSize: header.totalSize, }; this.loadedSamples.push(sample); this.nextTimestampInSamples += header.audioSamplesInFrame; - return; + return res.set(); } async getMimeType() { @@ -273,14 +282,14 @@ class Mp3AudioTrackBacking implements InputAudioTrackBacking { }; } - async getPacketAtIndex(sampleIndex: number, options: PacketRetrievalOptions) { + async getPacketAtIndex(res: ResultValue, sampleIndex: number, options: PacketRetrievalOptions): Promise { if (sampleIndex === -1) { - return null; + return res.set(null); } const rawSample = this.demuxer.loadedSamples[sampleIndex]; if (!rawSample) { - return null; + return res.set(null); } let data: Uint8Array; @@ -291,28 +300,33 @@ class Mp3AudioTrackBacking implements InputAudioTrackBacking { if (slice instanceof Promise) slice = await slice; if (!slice) { - return null; // Data didn't fit into the rest of the file + return res.set(null); // Data didn't fit into the rest of the file } data = readBytes(slice, rawSample.dataSize); } - return new EncodedPacket( + return res.set(new EncodedPacket( data, 'key', rawSample.timestamp, rawSample.duration, sampleIndex, rawSample.dataSize, - ); + )); } - getFirstPacket(options: PacketRetrievalOptions) { - return this.getPacketAtIndex(0, options); + getFirstPacket(res: ResultValue, options: PacketRetrievalOptions) { + return this.getPacketAtIndex(res, 0, options); } - async getNextPacket(packet: EncodedPacket, options: PacketRetrievalOptions) { - const release = await this.demuxer.readingMutex.acquire(); + async getNextPacket(res: ResultValue, packet: EncodedPacket, options: PacketRetrievalOptions): Promise { + // using foo = 5; + while (this.demuxer.readingMutex.locked) await this.demuxer.readingMutex.promise; + + // eslint-disable-next-line @typescript-eslint/no-unused-vars + using _ = this.demuxer.readingMutex.lock(); + // const release = await this.demuxer.readingMutex.acquire(); try { const sampleIndex = binarySearchExact( @@ -330,12 +344,14 @@ class Mp3AudioTrackBacking implements InputAudioTrackBacking { nextIndex >= this.demuxer.loadedSamples.length && !this.demuxer.lastSampleLoaded ) { - await this.demuxer.advanceReader(); + const result = new ResultValue(); + const promise = this.demuxer.advanceReader(result); + if (result.pending) await promise; } - return this.getPacketAtIndex(nextIndex, options); + return this.getPacketAtIndex(res, nextIndex, options); } finally { - release(); + // release(); } } diff --git a/src/mp3/mp3-reader.ts b/src/mp3/mp3-reader.ts index f489ca9..693b55c 100644 --- a/src/mp3/mp3-reader.ts +++ b/src/mp3/mp3-reader.ts @@ -7,12 +7,18 @@ */ import { FRAME_HEADER_SIZE, FrameHeader, readFrameHeader } from '../../shared/mp3-misc'; +import { ResultValue, Yo } from '../misc'; import { Reader, readU32Be } from '../reader'; -export const readNextFrameHeader = async (reader: Reader, startPos: number, until: number | null): Promise<{ - header: FrameHeader; - startPos: number; -} | null> => { +export const readNextFrameHeader = async ( + res: ResultValue<{ + header: FrameHeader; + startPos: number; + } | null>, + reader: Reader, + startPos: number, + until: number | null, +): Promise => { let currentPos = startPos; while (until === null || currentPos < until) { @@ -24,11 +30,11 @@ export const readNextFrameHeader = async (reader: Reader, startPos: number, unti const result = readFrameHeader(word, reader.fileSize !== null ? reader.fileSize - currentPos : null); if (result.header) { - return { header: result.header, startPos: currentPos }; + return res.set({ header: result.header, startPos: currentPos }); } currentPos += result.bytesAdvanced; } - return null; + return res.set(null); };