From c93755def1e3a0a592ca85be18428bae7e831ab3 Mon Sep 17 00:00:00 2001 From: Vanilagy <1696106+Vanilagy@users.noreply.github.com> Date: Fri, 29 Aug 2025 19:39:50 +0200 Subject: [PATCH] Implement ReadOrchestrator class, add highly optimized UrlSource --- src/adts/adts-demuxer.ts | 7 +- src/index.ts | 1 + src/input.ts | 7 +- src/isobmff/isobmff-demuxer.ts | 16 +- src/matroska/matroska-demuxer.ts | 24 +- src/misc.ts | 4 + src/mp3/mp3-demuxer.ts | 9 +- src/mp3/mp3-muxer.ts | 2 +- src/mp3/mp3-reader.ts | 5 +- src/ogg/ogg-demuxer.ts | 18 +- src/reader.ts | 205 ------ src/reader2.ts | 42 +- src/source.ts | 1118 ++++++++++++++++++------------ src/wave/riff-reader.ts | 63 -- src/wave/wave-demuxer.ts | 7 +- 15 files changed, 704 insertions(+), 824 deletions(-) delete mode 100644 src/reader.ts delete mode 100644 src/wave/riff-reader.ts diff --git a/src/adts/adts-demuxer.ts b/src/adts/adts-demuxer.ts index 61b8d82..438b1cd 100644 --- a/src/adts/adts-demuxer.ts +++ b/src/adts/adts-demuxer.ts @@ -44,7 +44,6 @@ export class AdtsDemuxer extends Demuxer { readingMutex = new AsyncMutex(); lastSampleLoaded = false; lastLoadedPos = 0; - fileSize = 0; nextTimestampInSamples = 0; constructor(input: Input) { @@ -55,10 +54,6 @@ export class AdtsDemuxer extends Demuxer { async readMetadata() { return this.metadataPromise ??= (async () => { - let fileSize = this.reader.requestSize(); - if (fileSize instanceof Promise) fileSize = await fileSize; - this.fileSize = fileSize; - // Keep loading until we find the first frame header while (!this.firstFrameHeader && !this.lastSampleLoaded) { await this.advanceReader(); @@ -86,7 +81,7 @@ export class AdtsDemuxer extends Demuxer { return; } - if (header.startPos + header.frameLength > this.fileSize) { + if (header.startPos + header.frameLength > this.reader.fileSize) { // Frame doesn't fit in the rest of the file this.lastSampleLoaded = true; return; diff --git a/src/index.ts b/src/index.ts index 9354d29..a218ddb 100644 --- a/src/index.ts +++ b/src/index.ts @@ -98,6 +98,7 @@ export { StreamSourceOptions, BlobSource, UrlSource, + UrlSource2, UrlSourceOptions, } from './source'; export { diff --git a/src/input.ts b/src/input.ts index de134df..6e2857e 100644 --- a/src/input.ts +++ b/src/input.ts @@ -9,7 +9,6 @@ import { Demuxer } from './demuxer'; import { InputFormat } from './input-format'; import { assert } from './misc'; -import { Reader } from './reader'; import { Reader2 } from './reader2'; import { Source } from './source'; @@ -34,13 +33,12 @@ export class Input { /** @internal */ _formats: InputFormat[]; /** @internal */ - _mainReader: Reader; - /** @internal */ _demuxerPromise: Promise | null = null; /** @internal */ _format: InputFormat | null = null; _reader2: Reader2; + _size!: number; constructor(options: InputOptions) { if (!options || typeof options !== 'object') { @@ -55,14 +53,13 @@ export class Input { this._formats = options.formats; this._source = options.source; - this._mainReader = new Reader(options.source); this._reader2 = new Reader2(options.source); } /** @internal */ _getDemuxer() { return this._demuxerPromise ??= (async () => { - await this._mainReader.loadRange(0, 4096); // Load the first 4 kiB so we can determine the format + this._reader2.fileSize = await this._source.getSize(); for (const format of this._formats) { const canRead = await format._canReadInput(this); diff --git a/src/isobmff/isobmff-demuxer.ts b/src/isobmff/isobmff-demuxer.ts index 6d002b4..4acda9d 100644 --- a/src/isobmff/isobmff-demuxer.ts +++ b/src/isobmff/isobmff-demuxer.ts @@ -271,11 +271,8 @@ export class IsobmffDemuxer extends Demuxer { readMetadata() { return this.metadataPromise ??= (async () => { - let sourceSize = this.reader.requestSize(); - if (sourceSize instanceof Promise) sourceSize = await sourceSize; - let currentPos = 0; - while (currentPos < sourceSize) { + while (currentPos < this.reader.fileSize) { let slice = this.reader.requestSliceRange(currentPos, MIN_BOX_HEADER_SIZE, MAX_BOX_HEADER_SIZE); if (slice instanceof Promise) slice = await slice; if (!slice) break; @@ -315,14 +312,14 @@ export class IsobmffDemuxer extends Demuxer { if (this.isFragmented) { // The last 4 bytes may contain the size of the mfra box at the end of the file - let lastWordSlice = this.reader.requestSlice(sourceSize - 4, 4); + let lastWordSlice = this.reader.requestSlice(this.reader.fileSize - 4, 4); if (lastWordSlice instanceof Promise) lastWordSlice = await lastWordSlice; assert(lastWordSlice); const lastWord = readU32Be(lastWordSlice); - const potentialMfraPos = sourceSize - lastWord; + const potentialMfraPos = this.reader.fileSize - lastWord; - if (potentialMfraPos >= 0 && potentialMfraPos <= sourceSize - MAX_BOX_HEADER_SIZE) { + if (potentialMfraPos >= 0 && potentialMfraPos <= this.reader.fileSize - MAX_BOX_HEADER_SIZE) { let mfraHeaderSlice = this.reader.requestSliceRange( potentialMfraPos, MIN_BOX_HEADER_SIZE, @@ -2417,9 +2414,6 @@ abstract class IsobmffTrackBacking implements InputTrackBacking { return this.fetchPacketInFragment(fragment, sampleIndex, options); } - let sourceSize = demuxer.reader.requestSize(); - if (sourceSize instanceof Promise) sourceSize = await sourceSize; - let prevFragment: Fragment | null = null; let bestFragmentIndex = fragmentIndex; let bestSampleIndex = sampleIndex; @@ -2455,7 +2449,7 @@ abstract class IsobmffTrackBacking implements InputTrackBacking { } } - while (currentPos < sourceSize) { + while (currentPos < demuxer.reader.fileSize) { if (prevFragment) { const trackData = prevFragment.trackData.get(this.internalTrack.id); if (trackData && trackData.startTimestamp > latestTimestamp) { diff --git a/src/matroska/matroska-demuxer.ts b/src/matroska/matroska-demuxer.ts index 4d66609..43b2503 100644 --- a/src/matroska/matroska-demuxer.ts +++ b/src/matroska/matroska-demuxer.ts @@ -234,12 +234,9 @@ export class MatroskaDemuxer extends Demuxer { readMetadata() { return this.readMetadataPromise ??= (async () => { - let fileSize = this.reader.requestSize(); - if (fileSize instanceof Promise) fileSize = await fileSize; - // Loop over all top-level elements in the file let currentPos = 0; - while (currentPos < fileSize) { + while (currentPos < this.reader.fileSize) { let slice = this.reader.requestSliceRange(currentPos, MIN_HEADER_SIZE, MAX_HEADER_SIZE); if (slice instanceof Promise) slice = await slice; if (!slice) break; @@ -281,9 +278,9 @@ export class MatroskaDemuxer extends Demuxer { this.reader, dataStartPos, LEVEL_0_AND_1_EBML_IDS, - fileSize, + this.reader.fileSize, ); - size = (nextElementPos ?? fileSize) - dataStartPos; + size = (nextElementPos ?? this.reader.fileSize) - dataStartPos; } const lastSegment = last(this.segments); @@ -422,12 +419,17 @@ export class MatroskaDemuxer extends Demuxer { } } - // Use the seek head to read missing metadata elements - for (const target of METADATA_ELEMENTS) { - if (this.currentSegment[target.flag]) continue; + // Sort the seek entries by file position so reading them exhibits a sequential pattern + this.currentSegment.seekEntries.sort((a, b) => a.segmentPosition - b.segmentPosition); - const seekEntry = this.currentSegment.seekEntries.find(entry => entry.id === target.id); - if (!seekEntry) continue; + // Use the seek head to read missing metadata elements + for (const seekEntry of this.currentSegment.seekEntries) { + const target = METADATA_ELEMENTS.find(x => x.id === seekEntry.id); + if (!target) { + continue; + } + + if (this.currentSegment[target.flag]) continue; let slice = this.reader.requestSliceRange( segmentDataStart + seekEntry.segmentPosition, diff --git a/src/misc.ts b/src/misc.ts index 94fdd65..7d2b1c8 100644 --- a/src/misc.ts +++ b/src/misc.ts @@ -633,3 +633,7 @@ export const isSafari = () => { * @public */ export type MaybePromise = T | Promise; + +export const closedIntervalsOverlap = (startA: number, endA: number, startB: number, endB: number) => { + return startA <= endB && startB <= endA; +}; diff --git a/src/mp3/mp3-demuxer.ts b/src/mp3/mp3-demuxer.ts index 3809d40..9d188dd 100644 --- a/src/mp3/mp3-demuxer.ts +++ b/src/mp3/mp3-demuxer.ts @@ -36,7 +36,6 @@ export class Mp3Demuxer extends Demuxer { readingMutex = new AsyncMutex(); lastSampleLoaded = false; lastLoadedPos = 0; - fileSize = 0; nextTimestampInSamples = 0; constructor(input: Input) { @@ -47,12 +46,8 @@ export class Mp3Demuxer extends Demuxer { async readMetadata() { return this.metadataPromise ??= (async () => { - let fileSize = this.reader.requestSize(); - if (fileSize instanceof Promise) fileSize = await fileSize; - this.fileSize = fileSize; - // Keep loading until we find the first frame header - while (!this.firstFrameHeader && this.lastLoadedPos < this.fileSize) { + while (!this.firstFrameHeader && this.lastLoadedPos < this.reader.fileSize) { await this.advanceReader(); } @@ -82,7 +77,7 @@ export class Mp3Demuxer extends Demuxer { const startPos = this.lastLoadedPos; - const result = await readNextFrameHeader(this.reader, startPos, this.fileSize); + const result = await readNextFrameHeader(this.reader, startPos, this.reader.fileSize); if (!result) { this.lastSampleLoaded = true; return; diff --git a/src/mp3/mp3-muxer.ts b/src/mp3/mp3-muxer.ts index 051bcf9..509c005 100644 --- a/src/mp3/mp3-muxer.ts +++ b/src/mp3/mp3-muxer.ts @@ -59,7 +59,7 @@ export class Mp3Muxer extends Muxer { } const word = view.getUint32(0, false); - const header = readFrameHeader(word, { pos: 0, fileSize: null }); + const header = readFrameHeader(word, null).header; if (!header) { throw new Error('Invalid MP3 header in sample.'); } diff --git a/src/mp3/mp3-reader.ts b/src/mp3/mp3-reader.ts index 40bff2d..90cf143 100644 --- a/src/mp3/mp3-reader.ts +++ b/src/mp3/mp3-reader.ts @@ -26,9 +26,6 @@ export const readNextFrameHeader = async (reader: Reader2, startPos: number, unt header: FrameHeader; startPos: number; } | null> => { - let fileSize = reader.requestSize(); - if (fileSize instanceof Promise) fileSize = await fileSize; - let currentPos = startPos; while (currentPos < until) { @@ -38,7 +35,7 @@ export const readNextFrameHeader = async (reader: Reader2, startPos: number, unt const word = readU32Be(slice); - const result = readFrameHeader(word, fileSize - currentPos); + const result = readFrameHeader(word, reader.fileSize - currentPos); if (result.header) { return { header: result.header, startPos: currentPos }; } diff --git a/src/ogg/ogg-demuxer.ts b/src/ogg/ogg-demuxer.ts index 3ba43f3..559deec 100644 --- a/src/ogg/ogg-demuxer.ts +++ b/src/ogg/ogg-demuxer.ts @@ -46,7 +46,6 @@ export class OggDemuxer extends Demuxer { reader: Reader2; metadataPromise: Promise | null = null; - fileSize: number | null = null; bitstreams: LogicalBitstream[] = []; tracks: InputAudioTrack[] = []; @@ -58,12 +57,8 @@ export class OggDemuxer extends Demuxer { async readMetadata() { return this.metadataPromise ??= (async () => { - let fileSize = this.reader.requestSize(); - if (fileSize instanceof Promise) fileSize = await fileSize; - this.fileSize = fileSize; - let currentPos = 0; - while (currentPos <= this.fileSize - MIN_PAGE_HEADER_SIZE) { + while (currentPos <= this.reader.fileSize - MIN_PAGE_HEADER_SIZE) { let slice = this.reader.requestSliceRange(currentPos, MIN_PAGE_HEADER_SIZE, MAX_PAGE_HEADER_SIZE); if (slice instanceof Promise) slice = await slice; if (!slice) break; @@ -249,7 +244,6 @@ export class OggDemuxer extends Demuxer { async readPacket(startPage: Page, startSegmentIndex: number): Promise { assert(startSegmentIndex < startPage.lacingValues.length); - assert(this.fileSize); let startDataOffset = 0; for (let i = 0; i < startSegmentIndex; i++) { @@ -290,7 +284,7 @@ export class OggDemuxer extends Demuxer { // The packet extends to the next page; let's find it let currentPos = currentPage.headerStartPos + currentPage.totalSize; while (true) { - if (currentPos > this.fileSize - MIN_PAGE_HEADER_SIZE) { + if (currentPos > this.reader.fileSize - MIN_PAGE_HEADER_SIZE) { return null; } @@ -335,8 +329,6 @@ export class OggDemuxer extends Demuxer { } async findNextPacketStart(lastPacket: Packet) { - assert(this.fileSize !== null); - // If there's another segment in the same page, return it if (lastPacket.endSegmentIndex < lastPacket.endPage.lacingValues.length - 1) { return { startPage: lastPacket.endPage, startSegmentIndex: lastPacket.endSegmentIndex + 1 }; @@ -351,7 +343,7 @@ export class OggDemuxer extends Demuxer { // Otherwise, search for the next page belonging to the same bitstream let currentPos = lastPacket.endPage.headerStartPos + lastPacket.endPage.totalSize; while (true) { - if (currentPos >= this.fileSize - MIN_PAGE_HEADER_SIZE) { + if (currentPos >= this.reader.fileSize - MIN_PAGE_HEADER_SIZE) { return null; } @@ -565,8 +557,6 @@ class OggAudioTrackBacking implements InputAudioTrackBacking { } async getPacket(timestamp: number, options: PacketRetrievalOptions) { - assert(this.demuxer.fileSize !== null); - const timestampInSamples = roundToPrecision(timestamp * this.internalSampleRate, 14); if (timestampInSamples === 0) { // Fast path for timestamp 0 - avoids binary search when playing back from the start @@ -584,7 +574,7 @@ class OggAudioTrackBacking implements InputAudioTrackBacking { } let lowPage = startPosition.startPage; - let high = this.demuxer.fileSize; + let high = this.demuxer.reader.fileSize; const lowPages: Page[] = [lowPage]; diff --git a/src/reader.ts b/src/reader.ts deleted file mode 100644 index 138601a..0000000 --- a/src/reader.ts +++ /dev/null @@ -1,205 +0,0 @@ -/*! - * Copyright (c) 2025-present, Vanilagy and contributors - * - * This Source Code Form is subject to the terms of the Mozilla Public - * License, v. 2.0. If a copy of the MPL was not distributed with this - * file, You can obtain one at https://mozilla.org/MPL/2.0/. - */ - -import { assert, binarySearchLessOrEqual, removeItem } from './misc'; -import { Source } from './source'; - -type ReadSegment = { - start: number; - end: number; - bytes: Uint8Array; - view: DataView; - age: number; -}; - -type LoadingSegment = { - start: number; - end: number; - promise: Promise; -}; - -export class Reader { - loadedSegments: ReadSegment[] = []; - loadingSegments: LoadingSegment[] = []; - sourceSizePromise: Promise | null = null; - nextAge = 0; - totalStoredBytes = 0; - - constructor(public source: Source, public maxStorableBytes = Infinity) {} - - async loadRange(start: number, end: number) { - end = Math.min(end, await this.source.getSize()); - - if (start >= end) { - return; - } - - const matchingLoadingSegment = this.loadingSegments.find(x => x.start <= start && x.end >= end); - if (matchingLoadingSegment) { - // Simply wait for the existing promise to finish to avoid loading the same range twice - await matchingLoadingSegment.promise; - return; - } - - const index = binarySearchLessOrEqual( - this.loadedSegments, - start, - x => x.start, - ); - if (index !== -1) { - for (let i = index; i < this.loadedSegments.length; i++) { - const segment = this.loadedSegments[i]!; - if (segment.start > start) { - break; - } - - const segmentEncasesRequestedRange = segment.end >= end; - if (segmentEncasesRequestedRange) { - // Nothing to load - return; - } - } - } - - this.source.onread?.(start, end); - const bytesPromise = this.source._read(start, end); - const loadingSegment: LoadingSegment = { start, end, promise: bytesPromise }; - this.loadingSegments.push(loadingSegment); - - const bytes = await bytesPromise; - removeItem(this.loadingSegments, loadingSegment); - - this.insertIntoLoadedSegments(start, bytes); - } - - rangeIsLoaded(start: number, end: number) { - if (end <= start) { - return true; - } - - const index = binarySearchLessOrEqual(this.loadedSegments, start, x => x.start); - if (index === -1) { - return false; - } - - for (let i = index; i < this.loadedSegments.length; i++) { - const segment = this.loadedSegments[i]!; - if (segment.start > start) { - break; - } - - const segmentEncasesRequestedRange = segment.end >= end; - if (segmentEncasesRequestedRange) { - return true; - } - } - - return false; - } - - private insertIntoLoadedSegments(start: number, bytes: Uint8Array) { - const segment: ReadSegment = { - start, - end: start + bytes.byteLength, - bytes, - view: new DataView(bytes.buffer), - age: this.nextAge++, - }; - - let index = binarySearchLessOrEqual(this.loadedSegments, start, x => x.start); - if (index === -1 || this.loadedSegments[index]!.start < segment.start) { - index++; - } - - // Insert the segment at the right place so that the array remains sorted by start offset - this.loadedSegments.splice(index, 0, segment); - this.totalStoredBytes += bytes.byteLength; - - // Remove all other segments from the array that are completely covered by the newly-inserted segment - for (let i = index + 1; i < this.loadedSegments.length; i++) { - const otherSegment = this.loadedSegments[i]!; - if (otherSegment.start >= segment.end) { - break; - } - - if (segment.start <= otherSegment.start && otherSegment.end <= segment.end) { - this.loadedSegments.splice(i, 1); - i--; - } - } - - // If we overshoot the max amount of permitted bytes, let's start evicting the oldest segments - while (this.totalStoredBytes > this.maxStorableBytes && this.loadedSegments.length > 1) { - let oldestSegment: ReadSegment | null = null; - let oldestSegmentIndex = -1; - - for (let i = 0; i < this.loadedSegments.length; i++) { - const candidate = this.loadedSegments[i]!; - if (!oldestSegment || candidate.age < oldestSegment.age) { - oldestSegment = candidate; - oldestSegmentIndex = i; - } - } - - assert(oldestSegment); - - this.totalStoredBytes -= oldestSegment.bytes.byteLength; - this.loadedSegments.splice(oldestSegmentIndex, 1); - } - } - - getViewAndOffset(start: number, end: number) { - const startIndex = binarySearchLessOrEqual(this.loadedSegments, start, x => x.start); - let segment: ReadSegment | null = null; - - if (startIndex !== -1) { - for (let i = startIndex; i < this.loadedSegments.length; i++) { - const candidate = this.loadedSegments[i]!; - - if (candidate.start > start) { - break; - } - - if (end <= candidate.end) { - segment = candidate; - break; - } - } - } - - if (!segment) { - throw new Error(`No segment loaded for range [${start}, ${end}).`); - } - - segment.age = this.nextAge++; - - return { - view: segment.view, - offset: segment.bytes.byteOffset + start - segment.start, - }; - } - - forgetRange(start: number, end: number) { - if (end <= start) { - return; - } - - const startIndex = binarySearchLessOrEqual(this.loadedSegments, start, x => x.start); - if (startIndex === -1) { - return; - } - - const segment = this.loadedSegments[startIndex]!; - if (segment.start !== start || segment.end !== end) { - return; - } - - this.loadedSegments.splice(startIndex, 1); - this.totalStoredBytes -= segment.bytes.byteLength; - } -} diff --git a/src/reader2.ts b/src/reader2.ts index 2440da5..ff000fd 100644 --- a/src/reader2.ts +++ b/src/reader2.ts @@ -56,28 +56,15 @@ export class FileSlice { } export class Reader2 { - private size: number | null = null; + fileSize!: number; - constructor(public source: Source) { - - } - - requestSize(): MaybePromise { - if (this.size !== null) { - return this.size; - } - - const size = this.source._retrieveSize2(); - if (size instanceof Promise) { - void size.then(x => this.size = x); - return size; - } else { - this.size = size; - return size; - } - } + constructor(public source: Source) {} requestSlice(start: number, length: number): MaybePromise { + if (start + length > this.fileSize) { + return null; + } + const end = start + length; const result = this.source._read2(start, end); @@ -99,19 +86,10 @@ export class Reader2 { } requestSliceRange(start: number, minLength: number, maxLength: number): MaybePromise { - const fileSize = this.requestSize(); - - if (fileSize instanceof Promise) { - return fileSize.then(size => this.requestSlice( - start, - clamp(size - start, minLength, maxLength), - )); - } else { - return this.requestSlice( - start, - clamp(fileSize - start, minLength, maxLength), - ); - } + return this.requestSlice( + start, + clamp(this.fileSize - start, minLength, maxLength), + ); } } diff --git a/src/source.ts b/src/source.ts index 7f21d98..cdb4e03 100644 --- a/src/source.ts +++ b/src/source.ts @@ -6,23 +6,27 @@ * file, You can obtain one at https://mozilla.org/MPL/2.0/. */ -import { assert, binarySearchLessOrEqual, insertSorted, MaybePromise, mergeObjectsDeeply, promiseWithResolvers, retriedFetch, toDataView } from './misc'; +import { + assert, + binarySearchLessOrEqual, + closedIntervalsOverlap, + MaybePromise, + mergeObjectsDeeply, + promiseWithResolvers, + retriedFetch, + toDataView, +} from './misc'; /** * The source base class, representing a resource from which bytes can be read. * @public */ export abstract class Source { - /** @internal */ - abstract _read(start: number, end: number): Promise; - /** @internal */ - abstract _retrieveSize(): Promise; - abstract _read2(start: number, end: number): MaybePromise<{ bytes: Uint8Array; view: DataView; offset: number; - } | null>; + }>; abstract _retrieveSize2(): MaybePromise; /** @internal */ @@ -32,8 +36,8 @@ export abstract class Source { * Resolves with the total size of the file in bytes. This function is memoized, meaning only the first call * will retrieve the size. */ - getSize() { - return this._sizePromise ??= this._retrieveSize(); + async getSize() { + return this._sizePromise ??= Promise.resolve(this._retrieveSize2()); } /** Called each time data is requested from the source. */ @@ -61,32 +65,16 @@ export class BufferSource extends Source { this._view = toDataView(this._bytes); } - /** @internal */ - async _read(start: number, end: number) { - return this._bytes.subarray(start, end); + _retrieveSize2() { + return this._bytes.byteLength; } - _read2(start: number, end: number) { - if (end > this._bytes.byteLength) { - return null; - } - + _read2() { return { bytes: this._bytes, view: this._view, offset: 0, }; - - // return this._bytes.subarray(start, end); - } - - _retrieveSize2() { - return this._bytes.byteLength; - } - - /** @internal */ - async _retrieveSize() { - return this._bytes.byteLength; } } @@ -136,33 +124,6 @@ export class StreamSource extends Source { } } -type BlobSourceReader = { - reader: ReadableStreamDefaultReader; - currentPos: number; - targetPos: number; - running: boolean; - pendingSlices: BlobSourcePendingSlice[]; - age: number; -}; - -type BlobSourceCacheEntry = { - start: number; - end: number; - bytes: Uint8Array; - view: DataView; - age: number; -}; - -type BlobSourcePendingSlice = { - start: number; - bytes: Uint8Array; - holes: { - start: number; - end: number; - }[]; - resolve: (bytes: Uint8Array) => void; -}; - /** * A source backed by a Blob. Since Files are also Blobs, this is the source to use when reading files off the disk. * @public @@ -170,10 +131,7 @@ type BlobSourcePendingSlice = { export class BlobSource extends Source { /** @internal */ _blob: Blob; - _cache: BlobSourceCacheEntry[] = []; - _totalCacheSize = 0; - _readers: BlobSourceReader[] = []; - _nextAge = 0; + _orchestrator: ReadOrchestrator; constructor(blob: Blob) { if (!(blob instanceof Blob)) { @@ -183,317 +141,57 @@ export class BlobSource extends Source { super(); this._blob = blob; + this._orchestrator = new ReadOrchestrator({ + maxCacheSize: 8 * 2 ** 20, // 8 MiB + maxWorkerCount: 4, + runWorker: this._runWorker.bind(this), + getPrefetchRange(start, end) { + const paddingStart = 2 ** 16; + const paddingEnd = 2 ** 17; + + start = Math.max(0, Math.floor((start - paddingStart) / paddingStart) * paddingStart); + end += paddingEnd; // Preload a tad into the future + + return { start, end }; + }, + }); + } + + _retrieveSize2() { + const size = this._blob.size; + this._orchestrator.fileSize = size; + + return size; } _read2(start: number, end: number) { - if (end > this._retrieveSize2()) { - return null; - } - - const cacheStartIndex = binarySearchLessOrEqual(this._cache, start, x => x.start); - const startEntry = cacheStartIndex !== -1 ? this._cache[cacheStartIndex] : null; - - if (startEntry && startEntry.start <= start && end <= startEntry.end) { - startEntry.age = this._nextAge++; - - return { - bytes: startEntry.bytes, - view: startEntry.view, - offset: startEntry.start, - }; - } - - const bytes = new Uint8Array(end - start); - let lastEnd = start; - const holes: { - start: number; - end: number; - }[] = []; - - if (cacheStartIndex !== -1) { - for (let i = cacheStartIndex; i < this._cache.length; i++) { - const entry = this._cache[i]!; - if (entry.start >= end) { - break; - } - if (entry.end <= start) { - continue; - } - - const cappedStart = Math.max(start, entry.start); - const cappedEnd = Math.min(end, entry.end); - assert(cappedStart <= cappedEnd); - - if (lastEnd < cappedStart) { - holes.push({ start: lastEnd, end: cappedStart }); - } - lastEnd = cappedEnd; - - bytes.set( - entry.bytes.subarray(cappedStart - entry.start, cappedEnd - entry.start), - cappedStart - start, - ); - entry.age = this._nextAge++; - } - - if (lastEnd < end) { - holes.push({ start: lastEnd, end }); - } - } else { - holes.push({ start, end }); - } - - if (holes.length === 0) { - return { - bytes, - view: toDataView(bytes), - offset: start, - }; - } - - const { promise, resolve } = promiseWithResolvers(); - - for (const hole of holes) { - const pendingSlice: BlobSourcePendingSlice = { - start, - bytes, - holes, - resolve, - }; - - const readerStart = Math.min(hole.start, Math.max(this._retrieveSize2() - 131072, 0)); - const readerEnd = hole.end;// Math.min(Math.max(readerStart + 131072, hole.end), this._retrieveSize2()); - - let readerFound = false; - for (const reader of this._readers) { - if (reader.currentPos <= readerStart && readerStart - 131072 <= reader.targetPos) { - reader.targetPos = Math.max(reader.targetPos, readerEnd); - readerFound = true; - - if (!reader.pendingSlices.includes(pendingSlice)) { - reader.pendingSlices.push(pendingSlice); - } - - if (!reader.running) { - void this._runReader(reader); - } - - break; - } - } - - if (!readerFound) { - const newReader: BlobSourceReader = { - reader: this._blob.slice(readerStart).stream().getReader(), - currentPos: readerStart, - targetPos: readerEnd, - running: false, - pendingSlices: [pendingSlice], - age: 0, // Will be set once we run it - }; - this._readers.push(newReader); - - void this._runReader(newReader); - - if (this._readers.length > 4) { - let oldestIndex = 0; - let oldestReader = this._readers[0]!; - - for (let i = 1; i < this._readers.length; i++) { - const reader = this._readers[i]!; - - if (reader.age < oldestReader.age) { - oldestIndex = i; - oldestReader = reader; - } - } - - this._readers.splice(oldestIndex, 1); - } - } - } - - return promise.then(bytes => ({ - bytes, - view: toDataView(bytes), - offset: start, - })); + return this._orchestrator.read(start, end); } - async _runReader(reader: BlobSourceReader) { - assert(!reader.running); - reader.running = true; - reader.age = this._nextAge++; + readers = new WeakMap>(); - while (reader.currentPos < reader.targetPos) { - const { done, value } = await reader.reader.read(); + async _runWorker(worker: ReadWorker) { + let reader = this.readers.get(worker); + if (!reader) { + // Get a reader of the blob starting at the required offset, and then keep it around + reader = this._blob.slice(worker.currentPos).stream().getReader(); + this.readers.set(worker, reader); + } + + while (worker.currentPos < worker.targetPos && !worker.aborted) { + const { done, value } = await reader.read(); if (done) { - const readerIndex = this._readers.indexOf(reader); - assert(readerIndex !== -1); - - this._readers.splice(readerIndex, 1); - + this._orchestrator.forgetWorker(worker); break; } - const start = reader.currentPos; - const end = start + value.length; - - this._insertIntoCache({ - start, - end, - bytes: value, - view: toDataView(value), - age: this._nextAge++, - }); - reader.currentPos += value.length; - reader.targetPos = Math.max(reader.targetPos, reader.currentPos); - - for (let i = 0; i < reader.pendingSlices.length; i++) { - const pendingSlice = reader.pendingSlices[i]!; - - const clampedStart = Math.max(start, pendingSlice.start); - const clampedEnd = Math.min(end, pendingSlice.start + pendingSlice.bytes.length); - - if (clampedStart < clampedEnd) { - pendingSlice.bytes.set( - value.subarray(clampedStart - start, clampedEnd - start), - clampedStart - pendingSlice.start, - ); - } - - for (let j = 0; j < pendingSlice.holes.length; j++) { - // The hole is intentionally not modified here if the read section starts somewhere in the middle of - // the hole. We don't need to do "hole splitting", since the readers are spawned *by* the holes, - // meaning there's always a reader which will consume the hole left to right. - const hole = pendingSlice.holes[j]!; - if (start <= hole.start && end > hole.start) { - hole.start = end; - } - - if (hole.end <= hole.start) { - pendingSlice.holes.splice(j, 1); - j--; - } - } - - if (pendingSlice.holes.length === 0) { - pendingSlice.resolve(pendingSlice.bytes); - reader.pendingSlices.splice(i, 1); - i--; - } - } + this._orchestrator.supplyWorkerData(worker, value); } - - reader.running = false; - } - - _insertIntoCache(entry: BlobSourceCacheEntry) { - let insertionIndex = binarySearchLessOrEqual(this._cache, entry.start, x => x.start) + 1; - - if (insertionIndex > 0) { - const previous = this._cache[insertionIndex - 1]!; - if (previous.end >= entry.end) { - // Previous entry swallows the one to be inserted; we don't need to do anything - return; - } - - if (previous.end > entry.start) { - // Partial overlap with the previous entry, let's join - const joined = new Uint8Array(entry.end - previous.start); - joined.set(previous.bytes, 0); - joined.set(entry.bytes, entry.start - previous.start); - - previous.bytes = joined; - previous.view = toDataView(joined); - previous.end = entry.end; - - // Do the rest of the logic with the previous entry instead - insertionIndex--; - entry = previous; - - this._totalCacheSize += entry.end - previous.end; - } else { - this._cache.splice(insertionIndex, 0, entry); - this._totalCacheSize += entry.bytes.length; - } - } else { - this._cache.splice(insertionIndex, 0, entry); - this._totalCacheSize += entry.bytes.length; - } - - for (let i = insertionIndex + 1; i < this._cache.length; i++) { - const next = this._cache[i]!; - if (entry.end <= next.start) { - // Even if they touch, we don't wanna merge them, no need - break; - } - - if (entry.end >= next.end) { - // The inserted entry completely swallows the next entry - this._cache.splice(i, 1); - this._totalCacheSize -= next.bytes.length; - i--; - continue; - } - - // Partial overlap, let's join - const joined = new Uint8Array(next.end - entry.start); - joined.set(entry.bytes, 0); - joined.set(next.bytes, next.start - entry.start); - - entry.bytes = joined; - entry.view = toDataView(joined); - entry.end = next.end; - this._cache.splice(i, 1); - - this._totalCacheSize -= entry.end - next.start; - - break; // After the join case, we're done: the next entry cannot possibly overlap with the inserted one. - } - - const MAX_CACHE_SIZE = 8 * 2 ** 20; // 8 MiB - - while (this._totalCacheSize > MAX_CACHE_SIZE) { - let oldestIndex = 0; - let oldestEntry = this._cache[0]!; - - for (let i = 1; i < this._cache.length; i++) { - const entry = this._cache[i]!; - - if (entry.age < oldestEntry.age) { - oldestIndex = i; - oldestEntry = entry; - } - } - - this._cache.splice(oldestIndex, 1); - this._totalCacheSize -= oldestEntry.bytes.length; - } - } - - _cachedSize: number | null = null; - _retrieveSize2() { - if (this._cachedSize !== null) { - return this._cachedSize; - } - - return this._cachedSize = this._blob.size; // Reading this field is expensive - } - - /** @internal */ - async _read(start: number, end: number) { - const slice = this._blob.slice(start, end); - const buffer = await slice.arrayBuffer(); - return new Uint8Array(buffer); - } - - /** @internal */ - async _retrieveSize() { - return this._blob.size; } } +const URL_SOURCE_MIN_LOAD_AMOUNT = 0.5 * 2 ** 20; // 0.5 MiB + /** * Options for UrlSource. * @public @@ -517,15 +215,15 @@ export type UrlSourceOptions = { * as it typically comes with increased latency. * @beta */ -export class UrlSource extends Source { - /** @internal */ - private _url: URL; - /** @internal */ - private _options: UrlSourceOptions; - /** @internal */ - private _fullData: ArrayBuffer | null = null; - /** @internal */ - private _nextUrlVersion: number | null = null; +export class UrlSource2 extends Source { + _url: URL; + _options: UrlSourceOptions; + _orchestrator: ReadOrchestrator; + + _existingResponses = new WeakMap(); constructor( url: string | URL, @@ -548,28 +246,78 @@ export class UrlSource extends Source { this._url = url instanceof URL ? url : new URL(url, location.href); this._options = options; + + this._orchestrator = new ReadOrchestrator({ + maxCacheSize: 64 * 2 ** 20, // 64 MiB + // Most files in the real-world have a single sequential access pattern, but having two in parallel can + // also happen + maxWorkerCount: 2, + runWorker: this._runWorker.bind(this), + getPrefetchRange(start, end, workers) { + // Add a slight bit of start padding because + const paddingStart = 2 ** 16; + start = Math.max(0, Math.floor((start - paddingStart) / paddingStart) * paddingStart); + + // Remote resources have extreme latency (relatively speaking), so the benefit from intelligent + // prefetching is great. The prefetch strategy employed for UrlSource is as follows: When we notice + // successive reads to a worker's read region, we prefetch more data at the end of that region, + // growing exponentially (up to a cap). This performs well for real-world use cases: Either we read a + // small part of the file once and then never need it again, in which case the requested about of data + // is small. Or, we're repeatedly doing a sequential access pattern (common in media files), in which + // case we can become more and more confident to prefetch more and more data. + for (const worker of workers) { + const maxExtensionAmount = 8 * 2 ** 20; // 8 MiB + + // When the read region cross the threshold point, we trigger a prefetch. This point is typically + // in the middle of the worker's read region, or a fixed offset from the end if the region has grown + // really large. + const thresholdPoint = Math.max( + (worker.startPos + worker.targetPos) / 2, + worker.targetPos - maxExtensionAmount, + ); + + if (closedIntervalsOverlap( + start, end, + thresholdPoint, worker.targetPos, + )) { + const size = worker.targetPos - worker.startPos; + + // If we extend by maxExtensionAmount + const a = Math.ceil((size + 1) / maxExtensionAmount) * maxExtensionAmount; + // If we extend to the next power of 2 + const b = 2 ** Math.ceil(Math.log2(size + 1)); + + const extent = Math.min(b, a); + end = Math.max(end, worker.startPos + extent); + } + } + + end = Math.max(end, start + URL_SOURCE_MIN_LOAD_AMOUNT); + + return { + start, + end, + }; + }, + }); } - /** @internal */ - private async _makeRequest( - range?: { start: number; end: number }, - ): Promise<{ response: ArrayBuffer; statusCode: number }> { - const headers: HeadersInit = {}; - - if (range) { - headers['Range'] = `bytes=${range.start}-${range.end - 1}`; - } - - if (this._nextUrlVersion !== null) { - this._url.searchParams.set('mediabunny_version', this._nextUrlVersion.toString()); - this._nextUrlVersion++; - } + async _retrieveSize2() { + // Retrieving the resource size for UrlSource is optimized: Almost always (= always), the first bytes we have to + // read are the start of the file. This means it's smart to combine size fetching with fetching the start of the + // file. We additionally use this step to probe if the server supports range requests, killing three birds with + // one stone. + const abortController = new AbortController(); const response = await retriedFetch( this._url, mergeObjectsDeeply(this._options.requestInit ?? {}, { - method: 'GET', - headers, + headers: { + // We could also send a non-range request to request the same bytes (all of them), but doing it like + // this is an easy way to check if the server supports range requests in the first place + Range: 'bytes=0-', + }, + signal: abortController.signal, }), this._options.getRetryDelay ?? (() => null), ); @@ -578,106 +326,556 @@ export class UrlSource extends Source { throw new Error(`Error fetching ${this._url}: ${response.status} ${response.statusText}`); } - const buffer = await response.arrayBuffer(); + let worker: ReadWorker; + let fileSize: number; - if ( - response.status === 206 - && range - && buffer.byteLength !== range.end - range.start - && this._nextUrlVersion === null - ) { - // We did a range request but it resolved with the wrong range; in Chromium, this can be due to a caching - // bug (https://issues.chromium.org/issues/436025873). Let's circumvent the cache for the rest of the - // session by appending a version to the URL. - this._nextUrlVersion = 1; - return this._makeRequest(range); + if (response.status === 206) { + fileSize = this._getPartialLengthFromRangeResponse(response); + worker = this._orchestrator.createWorker(0, URL_SOURCE_MIN_LOAD_AMOUNT); + } else { + // Server probably returned a 200. + + const contentLength = response.headers.get('Content-Length'); + if (contentLength) { + fileSize = Number(contentLength); + worker = this._orchestrator.createWorker(0, fileSize); + this._orchestrator.options.maxCacheSize = Infinity; // 🤷 + + console.warn( + 'HTTP server did not respond with 206 Partial Content, meaning the entire remote resource now has' + + ' to be downloaded. For efficient media file streaming across a network, please make sure your' + + ' server supports range requests.', + ); + } else { + throw new Error(`HTTP response (status ${response.status}) must surface Content-Length header.`); + } } - if (response.status === 200) { - // The server didn't return 206 Partial Content, so it's not a range response - this._fullData = buffer; - } + this._orchestrator.fileSize = fileSize; - return { - response: buffer, - statusCode: response.status, - }; + this._existingResponses.set(worker, { response, abortController }); + this._orchestrator.runWorker(worker); + + return fileSize; } - /** @internal */ - async _read(start: number, end: number): Promise { - if (this._fullData) { - return new Uint8Array(this._fullData, start, end - start); - } - - const { response, statusCode } = await this._makeRequest({ start, end }); - - // If server doesn't support range requests, it will return 200 instead of 206. In that case, let's manually - // slice the response. - if (statusCode === 200) { - const fullData = new Uint8Array(response); - return fullData.subarray(start, end); - } - - return new Uint8Array(response); + async _read2(start: number, end: number) { + return this._orchestrator.read(start, end); } - /** @internal */ - async _retrieveSize(): Promise { - if (this._fullData) { - return this._fullData.byteLength; - } + private async _runWorker(worker: ReadWorker) { + const existing = this._existingResponses.get(worker); - // First, try a HEAD request to get the size - try { - const headResponse = await retriedFetch( + let abortController = existing?.abortController; + let response = existing?.response; + + if (!abortController) { + abortController = new AbortController(); + response = await retriedFetch( this._url, mergeObjectsDeeply(this._options.requestInit ?? {}, { - method: 'HEAD', + headers: { + Range: `bytes=${worker.currentPos}-`, + }, + signal: abortController.signal, }), this._options.getRetryDelay ?? (() => null), ); - - if (headResponse.ok) { - const contentLength = headResponse.headers.get('Content-Length'); - if (contentLength) { - return parseInt(contentLength); - } - } - } catch { - // We tried } - // Try a range request to get the Content-Range header - const rangeResponse = await retriedFetch( - this._url, - mergeObjectsDeeply(this._options.requestInit ?? {}, { - method: 'GET', - headers: { Range: 'bytes=0-0' }, - }), - this._options.getRetryDelay ?? (() => null), - ); + assert(response); - if (rangeResponse.status === 206) { - const contentRange = rangeResponse.headers.get('Content-Range'); - if (contentRange) { - const match = contentRange.match(/bytes \d+-\d+\/(\d+)/); - if (match && match[1]) { - return parseInt(match[1]); - } + if (!response.ok) { + throw new Error(`Error fetching ${this._url}: ${response.status} ${response.statusText}`); + } + + const length = this._getPartialLengthFromRangeResponse(response); + const required = worker.targetPos - worker.currentPos; + if (length < required) { + throw new Error( + `HTTP response unexpectedly too short: Needed at least ${required} bytes, got only ${length}.`, + ); + } + + if (!response.body) { + throw new Error('Missing HTTP response body.'); + } + + const reader = response.body.getReader(); + + while (true) { + const { done, value } = await reader.read(); + if (done) { + this._orchestrator.forgetWorker(worker); + break; } - } else if (rangeResponse.status === 200) { - // The server just returned the whole thing - this._fullData = await rangeResponse.arrayBuffer(); - if (this._fullData.byteLength !== 1) { - return this._fullData.byteLength; + + this._orchestrator.supplyWorkerData(worker, value); + + if (worker.currentPos >= worker.targetPos || worker.aborted) { + abortController.abort(); + this._existingResponses.delete(worker); + break; + } + } + + // The previous UrlSource had logic for circumventing https://issues.chromium.org/issues/436025873; I haven't + // been able to observe this bug with the new UrlSource (maybe because we're using response streaming), so the + // logic for that has vanished for now. Leaving a comment here if this becomes relevant again. + } + + private _getPartialLengthFromRangeResponse(response: Response) { + const contentRange = response.headers.get('Content-Range'); + if (contentRange) { + const match = /\/(\d+)/.exec(contentRange); + if (match) { + return Number(match[1]); } else { - // The server responded with 200, but returned only the requested range, so skip the response + throw new Error(`Invalid Content-Range header: ${contentRange}`); + } + } else { + const contentLength = response.headers.get('Content-Length'); + if (contentLength) { + return Number(contentLength); + } else { + throw new Error( + 'Partial HTTP response (status 206) must surface either Content-Range or' + + ' Content-Length header.', + ); } } + } +} - // If the range request didn't provide the size, make a full GET request - const { response } = await this._makeRequest(); - return response.byteLength; +type PendingSlice = { + start: number; + bytes: Uint8Array; + holes: { + start: number; + end: number; + }[]; + resolve: (bytes: Uint8Array) => void; + reject: (error: unknown) => void; +}; + +type CacheEntry = { + start: number; + end: number; + bytes: Uint8Array; + view: DataView; + age: number; +}; + +type ReadWorker = { + startPos: number; + currentPos: number; + targetPos: number; + running: boolean; + aborted: boolean; + pendingSlices: PendingSlice[]; + age: number; +}; + +/** + * Godclass for orchestrating complex, cached read operations. The reading model is as follows: Any reading task is + * delegated to a *worker*, which is a sequential reader positioned somewhere along the file. All workers run in + * parallel and can be stopped and resumed in their forward movement. When read requests come in, this orchestrator will + * first try to satisfy the request with only the cached data. If this isn't possible, workers are spun up for all + * missing parts (or existing workers are repurposed), and these workers will then fill the holes in the data as they + * march along the file. + */ +class ReadOrchestrator { + fileSize: number | null = null; + nextAge = 0; // Used for LRU eviction of both cache entries and workers + workers: ReadWorker[] = []; + cache: CacheEntry[] = []; + currentCacheSize = 0; + + constructor(public options: { + maxCacheSize: number; + runWorker: (worker: ReadWorker) => Promise; + getPrefetchRange: (start: number, end: number, workers: ReadWorker[]) => { + start: number; + end: number; + }; + maxWorkerCount: number; + }) {} + + read(innerStart: number, innerEnd: number) { + assert(this.fileSize !== null); + + const prefetchRange = this.options.getPrefetchRange(innerStart, innerEnd, this.workers); + const outerStart = prefetchRange.start; + const outerEnd = Math.min(prefetchRange.end, this.fileSize); + assert(outerStart <= innerStart && innerEnd <= outerEnd); + + let result: MaybePromise<{ + bytes: Uint8Array; + view: DataView; + offset: number; + }> | null = null; + + const innerCacheStartIndex = binarySearchLessOrEqual(this.cache, innerStart, x => x.start); + const innerStartEntry = innerCacheStartIndex !== -1 ? this.cache[innerCacheStartIndex] : null; + + // See if the read request can be satisfied by a single cache entry + if (innerStartEntry && innerStartEntry.start <= innerStart && innerEnd <= innerStartEntry.end) { + innerStartEntry.age = this.nextAge++; + + result = { + bytes: innerStartEntry.bytes, + view: innerStartEntry.view, + offset: innerStartEntry.start, + }; + // Can't return yet though, still need to check if the prefetch range might lie outside the cached area + } + + const outerCacheStartIndex = binarySearchLessOrEqual(this.cache, outerStart, x => x.start); + + const bytes = result ? null : new Uint8Array(innerEnd - innerStart); + let contiguousBytesWriteEnd = 0; // Used to track if the cache is able to completely cover the bytes + + let lastEnd = outerStart; + // The "holes" in the cache (the parts we need to load) + const holes: { + start: number; + end: number; + }[] = []; + + // Loop over the cache and build up the list of holes + if (outerCacheStartIndex !== -1) { + for (let i = outerCacheStartIndex; i < this.cache.length; i++) { + const entry = this.cache[i]!; + if (entry.start >= outerEnd) { + break; + } + if (entry.end <= outerStart) { + continue; + } + + const cappedOuterStart = Math.max(outerStart, entry.start); + const cappedOuterEnd = Math.min(outerEnd, entry.end); + assert(cappedOuterStart <= cappedOuterEnd); + + if (lastEnd < cappedOuterStart) { + holes.push({ start: lastEnd, end: cappedOuterStart }); + } + lastEnd = cappedOuterEnd; + + if (bytes) { + const cappedInnerStart = Math.max(innerStart, entry.start); + const cappedInnerEnd = Math.min(innerEnd, entry.end); + + if (cappedInnerStart < cappedInnerEnd) { + const relativeOffset = cappedInnerStart - innerStart; + + // Fill the relevant section of the bytes with the cached data + bytes.set( + entry.bytes.subarray(cappedInnerStart - entry.start, cappedInnerEnd - entry.start), + relativeOffset, + ); + + if (relativeOffset === contiguousBytesWriteEnd) { + contiguousBytesWriteEnd = cappedInnerEnd - innerStart; + } + } + } + entry.age = this.nextAge++; + } + + if (lastEnd < outerEnd) { + holes.push({ start: lastEnd, end: outerEnd }); + } + } else { + holes.push({ start: outerStart, end: outerEnd }); + } + + if (bytes && contiguousBytesWriteEnd >= bytes.length) { + // Multiple cache entries were able to completely cover the requested bytes! + result = { + bytes, + view: toDataView(bytes), + offset: innerStart, + }; + } + + if (holes.length === 0) { + assert(result); + return result; + } + + // We need to read more data, so now we're in async land + const { promise, resolve, reject } = promiseWithResolvers(); + + // Fire off workers to take care of patching the holes + for (const hole of holes) { + const pendingSlice: PendingSlice | null = bytes && { + start: innerStart, + bytes, + holes, // Not yet correct! These are the outer holes, not the inner holes. Will be fixed further down! + resolve, + reject, + }; + + let workerFound = false; + for (const worker of this.workers) { + // A small tolerance in the case that the requested region is *just* after the target position of an + // existing worker. In that case, it's probably more efficient to repurpose that worker than to spawn + // another one so close to it + const gapCloserTolerance = 2 ** 17; + + if (closedIntervalsOverlap( + hole.start - gapCloserTolerance, hole.start, + worker.currentPos, worker.targetPos, + )) { + worker.targetPos = Math.max(worker.targetPos, hole.end); // Update the worker's target position + workerFound = true; + + if (pendingSlice && !worker.pendingSlices.includes(pendingSlice)) { + worker.pendingSlices.push(pendingSlice); + } + + if (!worker.running) { + // Kick it off if it's idle + this.runWorker(worker); + } + + break; + } + } + + if (!workerFound) { + // We need to spawn a new worker + const newWorker = this.createWorker(hole.start, hole.end); + if (pendingSlice) { + newWorker.pendingSlices = [pendingSlice]; + } + + this.runWorker(newWorker); + } + } + + // Turn the outer holes into inner holes + for (let i = 0; i < holes.length; i++) { + const hole = holes[i]!; + hole.start = Math.max(innerStart, hole.start); + hole.end = Math.min(innerEnd, hole.end); + + if (hole.end <= hole.start) { + // Empty hole + holes.splice(i, 1); + i--; + } + } + + if (!result) { + assert(bytes); + result = promise.then(bytes => ({ + bytes, + view: toDataView(bytes), + offset: innerStart, + })); + } else { + // The requested region was satisfied by the cache, but the entire prefetch region was not + } + + return result; + } + + createWorker(startPos: number, targetPos: number) { + const worker: ReadWorker = { + startPos, + currentPos: startPos, + targetPos, + running: false, + aborted: false, + pendingSlices: [], + age: this.nextAge++, + }; + this.workers.push(worker); + + // LRU eviction of the other workers + while (this.workers.length > this.options.maxWorkerCount) { + let oldestIndex = 0; + let oldestWorker = this.workers[0]!; + + for (let i = 1; i < this.workers.length; i++) { + const worker = this.workers[i]!; + + if (worker.age < oldestWorker.age) { + oldestIndex = i; + oldestWorker = worker; + } + } + + if (oldestWorker.running && oldestWorker.pendingSlices.length > 0) { + break; + } + + oldestWorker.aborted = true; + this.workers.splice(oldestIndex, 1); + } + + return worker; + } + + runWorker(worker: ReadWorker) { + assert(!worker.running); + assert(worker.currentPos < worker.targetPos); + + worker.running = true; + worker.age = this.nextAge++; + + void this.options.runWorker(worker) + .then(() => worker.running = false) + .catch((error) => { + if (worker.pendingSlices.length > 0) { + worker.pendingSlices.forEach(x => x.reject(error)); // Make sure to propagate any errors + } else { + throw error; // So it doesn't get swallowed + } + }); + } + + /** Called by a worker when it has read some data. */ + supplyWorkerData(worker: ReadWorker, bytes: Uint8Array) { + const start = worker.currentPos; + const end = start + bytes.length; + + this.insertIntoCache({ + start, + end, + bytes, + view: toDataView(bytes), + age: this.nextAge++, + }); + worker.currentPos += bytes.length; + + // Now, let's see if we can use the read bytes to fill any pending slice + for (let i = 0; i < worker.pendingSlices.length; i++) { + const pendingSlice = worker.pendingSlices[i]!; + + const clampedStart = Math.max(start, pendingSlice.start); + const clampedEnd = Math.min(end, pendingSlice.start + pendingSlice.bytes.length); + + if (clampedStart < clampedEnd) { + pendingSlice.bytes.set( + bytes.subarray(clampedStart - start, clampedEnd - start), + clampedStart - pendingSlice.start, + ); + } + + for (let j = 0; j < pendingSlice.holes.length; j++) { + // The hole is intentionally not modified here if the read section starts somewhere in the middle of + // the hole. We don't need to do "hole splitting", since the workers are spawned *by* the holes, + // meaning there's always a worker which will consume the hole left to right. + const hole = pendingSlice.holes[j]!; + if (start <= hole.start && end > hole.start) { + hole.start = end; + } + + if (hole.end <= hole.start) { + pendingSlice.holes.splice(j, 1); + j--; + } + } + + if (pendingSlice.holes.length === 0) { + // The slice has been fulfilled, everything has been read. Let's resolve the promise + pendingSlice.resolve(pendingSlice.bytes); + worker.pendingSlices.splice(i, 1); + i--; + } + } + } + + forgetWorker(worker: ReadWorker) { + const index = this.workers.indexOf(worker); + assert(index !== -1); + + this.workers.splice(index, 1); + } + + insertIntoCache(entry: CacheEntry) { + let insertionIndex = binarySearchLessOrEqual(this.cache, entry.start, x => x.start) + 1; + + if (insertionIndex > 0) { + const previous = this.cache[insertionIndex - 1]!; + if (previous.end >= entry.end) { + // Previous entry swallows the one to be inserted; we don't need to do anything + return; + } + + if (previous.end > entry.start) { + // Partial overlap with the previous entry, let's join + const joined = new Uint8Array(entry.end - previous.start); + joined.set(previous.bytes, 0); + joined.set(entry.bytes, entry.start - previous.start); + + previous.bytes = joined; + previous.view = toDataView(joined); + previous.end = entry.end; + + // Do the rest of the logic with the previous entry instead + insertionIndex--; + entry = previous; + + this.currentCacheSize += entry.end - previous.end; + } else { + this.cache.splice(insertionIndex, 0, entry); + this.currentCacheSize += entry.bytes.length; + } + } else { + this.cache.splice(insertionIndex, 0, entry); + this.currentCacheSize += entry.bytes.length; + } + + for (let i = insertionIndex + 1; i < this.cache.length; i++) { + const next = this.cache[i]!; + if (entry.end <= next.start) { + // Even if they touch, we don't wanna merge them, no need + break; + } + + if (entry.end >= next.end) { + // The inserted entry completely swallows the next entry + this.cache.splice(i, 1); + this.currentCacheSize -= next.bytes.length; + i--; + continue; + } + + // Partial overlap, let's join + const joined = new Uint8Array(next.end - entry.start); + joined.set(entry.bytes, 0); + joined.set(next.bytes, next.start - entry.start); + + entry.bytes = joined; + entry.view = toDataView(joined); + entry.end = next.end; + this.cache.splice(i, 1); + + this.currentCacheSize -= entry.end - next.start; + + break; // After the join case, we're done: the next entry cannot possibly overlap with the inserted one. + } + + // LRU eviction of cache entries + while (this.currentCacheSize > this.options.maxCacheSize && this.cache.length > 1) { + let oldestIndex = 0; + let oldestEntry = this.cache[0]!; + + for (let i = 1; i < this.cache.length; i++) { + const entry = this.cache[i]!; + + if (entry.age < oldestEntry.age) { + oldestIndex = i; + oldestEntry = entry; + } + } + + this.cache.splice(oldestIndex, 1); + this.currentCacheSize -= oldestEntry.bytes.length; + } } } diff --git a/src/wave/riff-reader.ts b/src/wave/riff-reader.ts deleted file mode 100644 index 21ccee4..0000000 --- a/src/wave/riff-reader.ts +++ /dev/null @@ -1,63 +0,0 @@ -/*! - * Copyright (c) 2025-present, Vanilagy and contributors - * - * This Source Code Form is subject to the terms of the Mozilla Public - * License, v. 2.0. If a copy of the MPL was not distributed with this - * file, You can obtain one at https://mozilla.org/MPL/2.0/. - */ - -import { Reader } from '../reader'; - -export class RiffReader { - pos = 0; - littleEndian = true; - - constructor(public reader: Reader) {} - - readBytes(length: number) { - const { view, offset } = this.reader.getViewAndOffset(this.pos, this.pos + length); - this.pos += length; - - return new Uint8Array(view.buffer, offset, length); - } - - readU16() { - const { view, offset } = this.reader.getViewAndOffset(this.pos, this.pos + 2); - this.pos += 2; - - return view.getUint16(offset, this.littleEndian); - } - - readU32() { - const { view, offset } = this.reader.getViewAndOffset(this.pos, this.pos + 4); - this.pos += 4; - - return view.getUint32(offset, this.littleEndian); - } - - readU64() { - let low: number; - let high: number; - - if (this.littleEndian) { - low = this.readU32(); - high = this.readU32(); - } else { - high = this.readU32(); - low = this.readU32(); - } - - return high * 0x100000000 + low; - } - - readAscii(length: number) { - const { view, offset } = this.reader.getViewAndOffset(this.pos, this.pos + length); - this.pos += length; - - let str = ''; - for (let i = 0; i < length; i++) { - str += String.fromCharCode(view.getUint8(offset + i)); - } - return str; - } -} diff --git a/src/wave/wave-demuxer.ts b/src/wave/wave-demuxer.ts index c576537..6b5f65a 100644 --- a/src/wave/wave-demuxer.ts +++ b/src/wave/wave-demuxer.ts @@ -47,9 +47,6 @@ export class WaveDemuxer extends Demuxer { async readMetadata() { return this.metadataPromise ??= (async () => { - let actualFileSize = this.reader.requestSize(); - if (actualFileSize instanceof Promise) actualFileSize = await actualFileSize; - let slice = this.reader.requestSlice(0, 12); if (slice instanceof Promise) slice = await slice; assert(slice); @@ -61,7 +58,7 @@ export class WaveDemuxer extends Demuxer { const outerChunkSize = readU32(slice, littleEndian); - let totalFileSize = isRf64 ? actualFileSize : Math.min(outerChunkSize + 8, actualFileSize); + let totalFileSize = isRf64 ? this.reader.fileSize : Math.min(outerChunkSize + 8, this.reader.fileSize); const format = readAscii(slice, 4); if (format !== 'WAVE') { @@ -98,7 +95,7 @@ export class WaveDemuxer extends Demuxer { const riffChunkSize = readU64(slice, littleEndian); dataChunkSize = readU64(slice, littleEndian); - totalFileSize = Math.min(riffChunkSize + 8, actualFileSize); + totalFileSize = Math.min(riffChunkSize + 8, this.reader.fileSize); } currentPos = startPos + chunkSize + (chunkSize & 1); // Handle padding