From 3bd227bd66bbfbf932a0264cd85cd69704eb5c73 Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Wed, 17 Dec 2025 17:44:52 +0100 Subject: [PATCH] Change pump reuse rule: enforce queued commands are resolved in the order in which they are queued --- src/cursors.ts | 98 ++++++++++++++---------------- test/browser/sample-cursor.test.ts | 68 +++++++++++++-------- 2 files changed, 89 insertions(+), 77 deletions(-) diff --git a/src/cursors.ts b/src/cursors.ts index eadb868..b3b8dfb 100644 --- a/src/cursors.ts +++ b/src/cursors.ts @@ -374,11 +374,11 @@ export abstract class SampleCursor< pumpStopQueued = false; pumpStopped = promiseWithResolvers(); - decodedTimestamps: number[] = []; - maxDecodedSequenceNumber = -1; pumpTarget: EncodedPacket | null = null; + lastTarget: EncodedPacket | null = null; pumpMutex = new AsyncMutex4(); _closed = false; + queuedResets = 0; error: unknown = null; errorSet = false; @@ -447,10 +447,6 @@ export abstract class SampleCursor< } } - while (this.decodedTimestamps.length > 0 && this.decodedTimestamps[0]! <= sample.timestamp) { - this.decodedTimestamps.shift(); - } - if (this.pendingRequests.length === 0) { if (this.pumpStopQueued) { sample.close(); @@ -525,6 +521,14 @@ export abstract class SampleCursor< } } + _ensureWillBeOpen() { + if (this.queuedResets > 0) { + return; + } + + this._ensureNotClosed(); + } + async _seekToPacket( res: ResultValue, targetPacketPromise: MaybePromise, @@ -565,9 +569,7 @@ export abstract class SampleCursor< let setNewPump = true; - if (this.sampleQueue.length > 0 && targetPacket.timestamp < this.sampleQueue[0]!.timestamp) { - - } else { + if (this.lastTarget && this.lastTarget.timestamp <= targetPacket.timestamp) { while (this.sampleQueue.length > 0) { const nextSample = this.sampleQueue.shift()!; this.queueDequeue.resolve(); @@ -581,31 +583,17 @@ export abstract class SampleCursor< } } - if (this.pumpTarget) { - const max = Math.max(this.pumpTarget.sequenceNumber, this.maxDecodedSequenceNumber); + if (targetPacket.timestamp - this.lastTarget.timestamp < 0.1) { + setNewPump = false; + } else { + let key = this.packetReader.readNextKey(this.lastTarget, { verifyKeyPackets: true }); + if (key instanceof Promise) key = await key; - if (targetPacket.sequenceNumber <= max) { - const nextExpectedTimestamp = this.decodedTimestamps[0] ?? this.packetCursor.current?.timestamp; - if (nextExpectedTimestamp === undefined || nextExpectedTimestamp > targetPacket.timestamp) { - // yeah - // formulate the "no nextExpectedTimestamp" case - } else { - setNewPump = false; - } - } else { - if (targetPacket.timestamp - this.pumpTarget.timestamp < 0.1) { - setNewPump = false; - } else { - let key = this.packetReader.readNextKey(this.pumpTarget, { verifyKeyPackets: true }); - if (key instanceof Promise) key = await key; - - if ( - !key - || targetPacket.sequenceNumber < key.sequenceNumber - ) { - setNewPump = false; - } - } + if ( + !key + || targetPacket.sequenceNumber < key.sequenceNumber + ) { + setNewPump = false; } } } @@ -614,6 +602,8 @@ export abstract class SampleCursor< await this.stopPump(); } + this.lastTarget = targetPacket; + if (!this.pumpTarget || targetPacket.sequenceNumber > this.pumpTarget.sequenceNumber) { this.pumpTarget = targetPacket; } @@ -648,7 +638,7 @@ export abstract class SampleCursor< } seekToFirst(): MaybePromise { - this._ensureNotClosed(); + this._ensureWillBeOpen(); try { const result = new ResultValue(); @@ -667,7 +657,7 @@ export abstract class SampleCursor< } seekTo(timestamp: number): MaybePromise { - this._ensureNotClosed(); + this._ensureWillBeOpen(); try { const result = new ResultValue(); @@ -686,7 +676,7 @@ export abstract class SampleCursor< } seekToKey(timestamp: number): MaybePromise { - this._ensureNotClosed(); + this._ensureWillBeOpen(); try { const result = new ResultValue(); @@ -751,21 +741,13 @@ export abstract class SampleCursor< } this.lastPendingRequest = pendingRequest; - // Note that the next packet we get here is not necessarily the packet belonging to the next sample, since we - // can have out of order timestamps when B-frames are at play. However, if next() is called a sufficiently - // large amount of times, then the pump target will stay roughly in sync with the desired next sample. - const next = this.pumpTarget && await this.packetReader.readNext(this.pumpTarget, { metadataOnly: true }); - if (next) { - this.pumpTarget = next; - } - lock.release(); // Waiting for the return would be too long return res.set(await request.promise); } next(): MaybePromise { - this._ensureNotClosed(); + this._ensureWillBeOpen(); try { const result = new ResultValue(); @@ -786,7 +768,7 @@ export abstract class SampleCursor< async iterate( callback: (sample: TransformedSample, stop: () => void) => MaybePromise, ) { - this._ensureNotClosed(); + this._ensureWillBeOpen(); let stopped = false; const stop = () => stopped = true; @@ -794,6 +776,12 @@ export abstract class SampleCursor< const waitPromise = this.waitUntilIdle(); if (waitPromise) await waitPromise; + const lock = this.pumpMutex.lock(); + if (lock.pending) await lock.ready; + lock.release(); + + this._ensureNotClosed(); + while (true) { if (this.current) { const result = callback(this.current, stop); @@ -815,11 +803,17 @@ export abstract class SampleCursor< // eslint-disable-next-line @stylistic/generator-star-spacing async *[Symbol.asyncIterator]() { - this._ensureNotClosed(); + this._ensureWillBeOpen(); const waitPromise = this.waitUntilIdle(); if (waitPromise) await waitPromise; + const lock = this.pumpMutex.lock(); + if (lock.pending) await lock.ready; + lock.release(); + + this._ensureNotClosed(); + while (true) { if (this.current) { yield this.current; @@ -932,8 +926,6 @@ export abstract class SampleCursor< continue; } - insertSorted(this.decodedTimestamps, this.packetCursor.current.timestamp, x => x); - this.maxDecodedSequenceNumber = this.packetCursor.current.sequenceNumber; this.decoder.decode(this.packetCursor.current); if (this.debugInfo.enabled) { @@ -974,10 +966,9 @@ export abstract class SampleCursor< this.pumpStopped.resolve(); this.pumpStopped = promiseWithResolvers(); this.pumpRunning = false; - this.maxDecodedSequenceNumber = -1; - this.decodedTimestamps.length = 0; this.pumpTarget = null; this.pumpStopQueued = false; + this.lastTarget = null; } } @@ -1009,6 +1000,9 @@ export abstract class SampleCursor< async reset() { this.predictedRequests++; + this.queuedResets++; + using _ = defer(() => this.queuedResets--); + using lock = this.pumpMutex.lock(); if (lock.pending) await lock.ready; @@ -1023,8 +1017,6 @@ export abstract class SampleCursor< assert(this.sampleQueue.length === 0); assert(this.pendingRequests.length === 0); assert(this.lastPendingRequest === null); - assert(this.decodedTimestamps.length === 0); - assert(this.maxDecodedSequenceNumber === -1); assert(this.pumpTarget === null); assert(!this.decoder || this.decoder.closed); diff --git a/test/browser/sample-cursor.test.ts b/test/browser/sample-cursor.test.ts index 2e8dea9..98c5cb8 100644 --- a/test/browser/sample-cursor.test.ts +++ b/test/browser/sample-cursor.test.ts @@ -10,7 +10,36 @@ import { VideoSampleCursor, } from '../../src/cursors.js'; import { AudioSample, VideoSample } from '../../src/sample.js'; -import { promiseIterateAll } from '../../src/misc.js'; +import { promiseIterateAll, promiseWithResolvers } from '../../src/misc.js'; + +const promiseAllEnsureOrder = async (promises: T[]) => { + const results: Awaited[] = []; + const { promise, resolve, reject } = promiseWithResolvers(); + + const onValue = (value: Awaited, i: number) => { + if (results.length === i) { + results.push(value); + + if (results.length === promises.length) { + resolve(); + } + } else { + reject(new Error('Order violation')); + } + }; + + for (let i = 0; i < promises.length; i++) { + const value = promises[i]!; + if (value instanceof Promise) { + void value.then(x => onValue(x as Awaited, i)); + } else { + onValue(value as Awaited, i); + } + } + + await promise; + return results; +}; test('Sample cursor seeking', async () => { using input = new Input({ @@ -392,8 +421,7 @@ test('Sample cursor reset', async () => { cursor.next(), ]; - // eslint-disable-next-line @typescript-eslint/await-thenable - const results = await Promise.all(commands); + const results = await promiseAllEnsureOrder(commands); expect(cursor.closed).toBe(false); expect(results[0]!.timestamp).toBeGreaterThan(firstSample!.timestamp); @@ -410,8 +438,7 @@ test('Sample cursor reset', async () => { cursor2.reset(), ]; - // eslint-disable-next-line @typescript-eslint/await-thenable - await Promise.all(commands2); + await promiseAllEnsureOrder(commands2); expect(cursor2.debugInfo.decodedPackets).toHaveLength(1); }); @@ -588,8 +615,7 @@ test('Command queuing', async () => { ]; expect(commands0.every(x => x instanceof Promise)).toBe(true); - // eslint-disable-next-line @typescript-eslint/await-thenable - await Promise.all(commands0); + await promiseAllEnsureOrder(commands0); expect(cursor0.debugInfo.pumpsStarted).toBe(1); @@ -602,8 +628,7 @@ test('Command queuing', async () => { ]; expect(commands1.every(x => x instanceof Promise)).toBe(true); - // eslint-disable-next-line @typescript-eslint/await-thenable - const results1 = await Promise.all(commands1); + const results1 = await promiseAllEnsureOrder(commands1); expect(results1[0]!.timestamp).toBe(0); expect(cursor1.debugInfo.decodedPackets.map(x => x.timestamp)).toEqual([0]); @@ -621,8 +646,7 @@ test('Command queuing', async () => { ]; expect(commands2.every(x => x instanceof Promise)).toBe(true); - // eslint-disable-next-line @typescript-eslint/await-thenable - const results2 = await Promise.all(commands2); + const results2 = await promiseAllEnsureOrder(commands2); expect(results2[0]!.timestamp).toBe(0); expect(results2[1]!.timestamp).toBe(1); expect(results2[2]!.timestamp).toBe(2); @@ -632,6 +656,7 @@ test('Command queuing', async () => { expect(cursor2.debugInfo.decodedPackets.map(x => x.timestamp)).toEqual([ 0, 1, 2, 3, 4, 5, ]); + expect(cursor2.debugInfo.pumpsStarted).toBe(6); const cursor3 = new VideoSampleCursor(reader); cursor3.debugInfo.enabled = true; @@ -647,8 +672,7 @@ test('Command queuing', async () => { ]; expect(commands3.every(x => x instanceof Promise)).toBe(true); - // eslint-disable-next-line @typescript-eslint/await-thenable - const results3 = await Promise.all(commands3); + const results3 = await promiseAllEnsureOrder(commands3); expect(results3[0]!.timestamp).toBeLessThanOrEqual(0.5); expect(results3[1]!.timestamp).toBeLessThanOrEqual(0.4); @@ -657,7 +681,7 @@ test('Command queuing', async () => { expect(results3[4]!.timestamp).toBeLessThanOrEqual(0.1); expect(results3[5]!.timestamp).toBe(0); expect(cursor3.debugInfo.decodedPackets.every(x => x.timestamp <= 0.5)).toBe(true); - expect(cursor3.debugInfo.pumpsStarted).toBe(1); + expect(cursor3.debugInfo.pumpsStarted).toBe(6); const cursor4 = new VideoSampleCursor(reader); cursor4.debugInfo.enabled = true; @@ -671,8 +695,7 @@ test('Command queuing', async () => { expect(commands4.every(x => x instanceof Promise)).toBe(true); - // eslint-disable-next-line @typescript-eslint/await-thenable - const results4 = await Promise.all(commands4); + const results4 = await promiseAllEnsureOrder(commands4); expect(results4[0]!.timestamp).toBe(0); expect(results4[1]!.timestamp).toBeGreaterThan(results4[0]!.timestamp); @@ -697,8 +720,7 @@ test('Command queuing', async () => { cursor5.close(), ]; - // eslint-disable-next-line @typescript-eslint/await-thenable - const results5 = await Promise.all(commands5); + const results5 = await promiseAllEnsureOrder(commands5); expect(results5[0]!.timestamp).toBe(0); expect(results5[1]!.timestamp).toBeGreaterThan(results5[0]!.timestamp); @@ -714,7 +736,7 @@ test('Command queuing', async () => { expect(results5[9]!.timestamp).toBe(0); expect(results5[10]!.timestamp).toBeGreaterThan(results5[9]!.timestamp); - expect(cursor5.debugInfo.pumpsStarted).toBe(1); + expect(cursor5.debugInfo.pumpsStarted).toBe(2); const cursor6 = new VideoSampleCursor(reader); cursor6.debugInfo.enabled = true; @@ -729,8 +751,7 @@ test('Command queuing', async () => { cursor6.close(), ]; - // eslint-disable-next-line @typescript-eslint/await-thenable - const results6 = await Promise.all(commands6); + const results6 = await promiseAllEnsureOrder(commands6); expect(results6[0]!.timestamp).toBe(0); expect(results6[1]!.timestamp).toBeLessThanOrEqual(0.4); @@ -748,7 +769,7 @@ test('Command queuing', async () => { ]; expect(commands7[0]).toBe(commands7[1]); // Same Promise - await Promise.all(commands7); + await promiseAllEnsureOrder(commands7); const cursor8 = new VideoSampleCursor(reader, { autoClose: false }); @@ -881,8 +902,7 @@ test('AudioSampleCursor', async () => { cursor.seekTo(0.2), ]; - // eslint-disable-next-line @typescript-eslint/await-thenable - const result = await Promise.all(commands); + const result = await promiseAllEnsureOrder(commands); expect(result[0]!.timestamp).toBeLessThanOrEqual(0); expect(result[1]!.timestamp).toBeLessThanOrEqual(0.05);