diff --git a/android/app/proguard-rules.pro b/android/app/proguard-rules.pro index fe1f0eb7d..2a20311e7 100644 --- a/android/app/proguard-rules.pro +++ b/android/app/proguard-rules.pro @@ -20,11 +20,12 @@ -keep class androidx.media3.decoder.SimpleDecoderOutputBuffer { *; } # ffmpeg_demuxer_jni.cc resolves the AVIO input proxy's callbacks by name -# (FindClass on the interface, GetMethodID for position/read/length). Keeping -# the interface members pins the names on every implementation, including the -# anonymous proxy inside FfmpegExtractor. +# (FindClass on the interface, GetMethodID for position/read/readAt/length). +# Keeping the interface members pins the names on every implementation, +# including the anonymous proxy inside FfmpegExtractor. -keep interface com.edde746.plezy.exoplayer.FfmpegDemuxerJni$Input { long position(); int read(byte[], int); + int readAt(long, byte[], int); long length(); } diff --git a/android/app/src/androidTest/assets/ffmpeg/seek_cued.mkv b/android/app/src/androidTest/assets/ffmpeg/seek_cued.mkv new file mode 100644 index 000000000..776cdc3c7 Binary files /dev/null and b/android/app/src/androidTest/assets/ffmpeg/seek_cued.mkv differ diff --git a/android/app/src/androidTest/assets/ffmpeg/seek_cueless.mkv b/android/app/src/androidTest/assets/ffmpeg/seek_cueless.mkv new file mode 100644 index 000000000..85ce73566 Binary files /dev/null and b/android/app/src/androidTest/assets/ffmpeg/seek_cueless.mkv differ diff --git a/android/app/src/androidTest/kotlin/com/edde746/plezy/exoplayer/FfmpegExtractorAudioCsdPlaybackTest.kt b/android/app/src/androidTest/kotlin/com/edde746/plezy/exoplayer/FfmpegExtractorAudioCsdPlaybackTest.kt index d37792c76..3acebb25c 100644 --- a/android/app/src/androidTest/kotlin/com/edde746/plezy/exoplayer/FfmpegExtractorAudioCsdPlaybackTest.kt +++ b/android/app/src/androidTest/kotlin/com/edde746/plezy/exoplayer/FfmpegExtractorAudioCsdPlaybackTest.kt @@ -7,6 +7,7 @@ import android.os.HandlerThread import androidx.media3.common.MediaItem import androidx.media3.common.PlaybackException import androidx.media3.common.Player +import androidx.media3.datasource.DataSpec import androidx.media3.datasource.DefaultDataSource import androidx.media3.exoplayer.ExoPlayer import androidx.media3.exoplayer.source.ProgressiveMediaSource @@ -66,12 +67,14 @@ class FfmpegExtractorAudioCsdPlaybackTest { handler.post { try { + val fixtureUri = Uri.fromFile(fixtureFile) val ffmpegOnlyExtractors = ExtractorsFactory { val extractor = FfmpegExtractor.create( { FfmpegDemuxerPolicy.Preference.FFMPEG }, DvConversionMode.DISABLED, DefaultSubtitleParserFactory(), - AssHandler() + AssHandler(), + FfmpegRandomAccessSource(DefaultDataSource.Factory(context)) { DataSpec(fixtureUri) } ) assertNotNull("FFmpeg demuxer JNI library is unavailable", extractor) arrayOf(extractor!!) @@ -95,7 +98,7 @@ class FfmpegExtractorAudioCsdPlaybackTest { val source = ProgressiveMediaSource.Factory( DefaultDataSource.Factory(context), ffmpegOnlyExtractors - ).createMediaSource(MediaItem.fromUri(Uri.fromFile(fixtureFile))) + ).createMediaSource(MediaItem.fromUri(fixtureUri)) player.setMediaSource(source) player.prepare() player.play() diff --git a/android/app/src/androidTest/kotlin/com/edde746/plezy/exoplayer/FfmpegExtractorSeekTest.kt b/android/app/src/androidTest/kotlin/com/edde746/plezy/exoplayer/FfmpegExtractorSeekTest.kt new file mode 100644 index 000000000..6e21f441f --- /dev/null +++ b/android/app/src/androidTest/kotlin/com/edde746/plezy/exoplayer/FfmpegExtractorSeekTest.kt @@ -0,0 +1,321 @@ +package com.edde746.plezy.exoplayer + +import android.content.Context +import android.net.Uri +import androidx.media3.common.C +import androidx.media3.common.DataReader +import androidx.media3.common.Format +import androidx.media3.common.util.ParsableByteArray +import androidx.media3.datasource.DataSource +import androidx.media3.datasource.DataSpec +import androidx.media3.datasource.DefaultDataSource +import androidx.media3.datasource.TransferListener +import androidx.media3.extractor.DefaultExtractorInput +import androidx.media3.extractor.Extractor +import androidx.media3.extractor.ExtractorInput +import androidx.media3.extractor.ExtractorOutput +import androidx.media3.extractor.PositionHolder +import androidx.media3.extractor.SeekMap +import androidx.media3.extractor.TrackOutput +import androidx.media3.extractor.text.DefaultSubtitleParserFactory +import androidx.test.ext.junit.runners.AndroidJUnit4 +import androidx.test.platform.app.InstrumentationRegistry +import com.edde746.plezy.libass.media.AssHandler +import java.io.File +import org.junit.Assert.assertEquals +import org.junit.Assert.assertTrue +import org.junit.Test +import org.junit.runner.RunWith + +/** + * Regression coverage for #2096 and for the seam that caused it. + * + * libavformat resolves seeks through its own index, reading it via + * [FfmpegRandomAccessSource]. When that read cannot be served, matroskadec + * burns its one-shot Cues parse and every later seek degrades to libavformat's + * linear generic scan — which is what buffered a resume for minutes. The cued + * fixture here has the same geometry as the file from the report (front + * SeekHead, Cues at 99.5% of the file), so a scan is trivially distinguishable + * from an index seek: it has to read most of the file. + */ +@RunWith(AndroidJUnit4::class) +class FfmpegExtractorSeekTest { + + private companion object { + // Regenerate with: + // ffmpeg -f lavfi -i "testsrc=size=160x120:rate=10:duration=30" \ + // -c:v libx264 -g 10 -pix_fmt yuv420p -preset veryfast seek_cued.mkv + // mkvmerge --no-cues -o seek_cueless.mkv seek_cued.mkv + // The cued fixture keeps ffmpeg's default layout: a SeekHead at byte 52 + // pointing at Cues written after the clusters, at 99.5% of the file. + const val CUED = "ffmpeg/seek_cued.mkv" + const val CUELESS = "ffmpeg/seek_cueless.mkv" + + /** The fixtures are 30 s at 10 fps with a keyframe every second. */ + const val TARGET_US = 25_000_000L + const val KEYFRAME_TOLERANCE_US = 1_500_000L + } + + @Test + fun cuedSeekLandsOnTheTargetWithoutWalkingTheFile() { + withHarness(CUED) { harness -> + harness.pumpUntilSampleDelivered() + assertTrue( + "playback should start at the beginning, got ${harness.lastSampleUs}us", + harness.lastSampleUs < 500_000L + ) + + harness.resetByteCounter() + harness.seekTo(TARGET_US) + val landed = harness.pumpUntilSampleAtLeast(TARGET_US - KEYFRAME_TOLERANCE_US) + + assertTrue("no sample delivered after seeking to ${TARGET_US}us", landed) + assertTrue( + "seek landed at ${harness.lastSampleUs}us, expected within " + + "${KEYFRAME_TOLERANCE_US}us of ${TARGET_US}us", + harness.lastSampleUs <= TARGET_US + KEYFRAME_TOLERANCE_US + ) + // An index seek reads the Cues plus the target cluster. A generic scan + // has to read everything up to 83% of the file to get here. + val budget = harness.fixtureSize / 3 + assertTrue( + "seek read ${harness.bytesRead} bytes of a ${harness.fixtureSize}-byte file; " + + "expected under $budget (index seek, not a scan)", + harness.bytesRead < budget + ) + // Reopening per read would mean one request per 64 KB in steady state. + assertTrue("random-access opens after seek: ${harness.randomAccessOpens}", harness.randomAccessOpens <= 4) + } + } + + /** + * Without Cues libavformat falls back to its generic scan, exactly as media3's + * own MatroskaExtractor scans clusters. That is slow but it must still land on + * the target rather than stalling or silently delivering from the wrong place. + */ + @Test + fun cuelessSeekStillReachesTheTarget() { + withHarness(CUELESS) { harness -> + harness.pumpUntilSampleDelivered() + harness.seekTo(TARGET_US) + val landed = harness.pumpUntilSampleAtLeast(TARGET_US - KEYFRAME_TOLERANCE_US) + + assertTrue("cueless seek to ${TARGET_US}us delivered nothing", landed) + assertTrue( + "cueless seek landed at ${harness.lastSampleUs}us", + harness.lastSampleUs <= TARGET_US + KEYFRAME_TOLERANCE_US + ) + } + } + + @Test + fun playbackFromTheStartDeliversEverySampleFromTheBeginning() { + withHarness(CUED) { harness -> + val timestamps = harness.pumpSamples(count = 40) + + assertTrue("expected samples from the start, got ${timestamps.size}", timestamps.size >= 40) + assertTrue("first sample at ${timestamps.first()}us", timestamps.first() < 500_000L) + // Delivery is in decode order, which is not presentation order once the + // encoder emits B-frames; what must hold is that the set of samples is + // gapless from zero (this fixture is a steady 10 fps). + val sorted = timestamps.sorted() + assertEquals(sorted.indices.map { it * 100_000L }, sorted) + } + } + + private fun withHarness(fixture: String, block: (Harness) -> Unit) { + val instrumentation = InstrumentationRegistry.getInstrumentation() + val context = instrumentation.targetContext + val file = copyFixture(instrumentation.context, context, fixture) + val harness = Harness(context, file) + try { + harness.prepare() + block(harness) + } finally { + harness.close() + file.delete() + } + } + + private fun copyFixture(instrumentationContext: Context, targetContext: Context, fixture: String): File { + val output = File.createTempFile("ffmpeg-seek-fixture-", ".mkv", targetContext.cacheDir) + instrumentationContext.assets.open(fixture).use { input -> + output.outputStream().use(input::copyTo) + } + return output + } + + /** + * Minimal stand-in for ProgressiveMediaPeriod's loader: it opens the data + * source, feeds an ExtractorInput, and honours RESULT_SEEK by re-opening — + * the same contract media3 implements in ExtractingLoadable.load(). + */ + private class Harness(context: Context, private val fixture: File) { + val fixtureSize = fixture.length() + private val uri: Uri = Uri.fromFile(fixture) + private val counting = CountingDataSourceFactory(DefaultDataSource.Factory(context)) + private val io = FfmpegRandomAccessSource(counting) { DataSpec(uri) } + private val output = CapturingOutput() + private val positionHolder = PositionHolder() + private val extractor = requireNotNull( + FfmpegExtractor.create( + { FfmpegDemuxerPolicy.Preference.FFMPEG }, + DvConversionMode.DISABLED, + DefaultSubtitleParserFactory(), + AssHandler(), + io + ) + ) { "native ffmpeg demuxer unavailable" } + + private var dataSource: DataSource? = null + private var input: ExtractorInput? = null + private var opensBaseline = 0 + + val lastSampleUs: Long + get() = output.lastSampleUs + + val bytesRead: Long + get() = counting.bytesRead + + val randomAccessOpens: Int + get() = io.openCount - opensBaseline + + fun prepare() { + openAt(0) + assertTrue("fixture should be sniffed by the ffmpeg demuxer", extractor.sniff(input!!)) + input!!.resetPeekPosition() + extractor.init(output) + } + + fun resetByteCounter() { + counting.reset() + opensBaseline = io.openCount + } + + fun seekTo(timeUs: Long) { + // media3 restarts the load at the SeekMap's byte hint, then calls seek() + // on the loader thread before the next read. + val hint = output.seekMap?.getSeekPoints(timeUs)?.first?.position ?: 0L + openAt(hint) + extractor.seek(hint, timeUs) + } + + fun pumpUntilSampleDelivered(): Boolean { + val before = output.sampleCount + return pump { output.sampleCount > before } + } + + fun pumpUntilSampleAtLeast(timeUs: Long): Boolean = pump { output.lastSampleUs >= timeUs } + + fun pumpSamples(count: Int): List { + pump { output.sampleTimestamps.size >= count } + return output.sampleTimestamps.toList() + } + + private fun pump(until: () -> Boolean): Boolean { + repeat(200_000) { + if (until()) return true + when (extractor.read(input!!, positionHolder)) { + Extractor.RESULT_SEEK -> openAt(positionHolder.position) + Extractor.RESULT_END_OF_INPUT -> return until() + else -> Unit + } + } + return until() + } + + private fun openAt(position: Long) { + dataSource?.close() + val source = counting.createDataSource() + val remaining = source.open(DataSpec.Builder().setUri(uri).setPosition(position).build()) + dataSource = source + input = DefaultExtractorInput( + source, + position, + if (remaining == C.LENGTH_UNSET.toLong()) C.LENGTH_UNSET.toLong() else position + remaining + ) + } + + fun close() { + extractor.release() + dataSource?.close() + } + } + + private class CapturingOutput : ExtractorOutput { + val sampleTimestamps = mutableListOf() + var seekMap: SeekMap? = null + private set + + val sampleCount: Int + get() = sampleTimestamps.size + + val lastSampleUs: Long + get() = sampleTimestamps.lastOrNull() ?: C.TIME_UNSET + + override fun track(id: Int, type: Int): TrackOutput = CapturingTrack(type, sampleTimestamps) + + override fun endTracks() = Unit + + override fun seekMap(seekMap: SeekMap) { + this.seekMap = seekMap + } + } + + private class CapturingTrack(private val type: Int, private val timestamps: MutableList) : TrackOutput { + private val scratch = ByteArray(64 * 1024) + + override fun format(format: Format) = Unit + + override fun sampleData(input: DataReader, length: Int, allowEndOfInput: Boolean): Int = sampleData(input, length, allowEndOfInput, TrackOutput.SAMPLE_DATA_PART_MAIN) + + override fun sampleData(input: DataReader, length: Int, allowEndOfInput: Boolean, sampleDataPart: Int): Int { + val read = input.read(scratch, 0, minOf(length, scratch.size)) + return if (read == C.RESULT_END_OF_INPUT) 0 else read + } + + override fun sampleData(data: ParsableByteArray, length: Int) { + sampleData(data, length, TrackOutput.SAMPLE_DATA_PART_MAIN) + } + + override fun sampleData(data: ParsableByteArray, length: Int, sampleDataPart: Int) { + data.skipBytes(length) + } + + override fun sampleMetadata(timeUs: Long, flags: Int, size: Int, offset: Int, cryptoData: TrackOutput.CryptoData?) { + if (type == C.TRACK_TYPE_VIDEO) timestamps.add(timeUs) + } + } + + /** Counts every byte the demuxer pulls, through either read path. */ + private class CountingDataSourceFactory(private val delegate: DataSource.Factory) : DataSource.Factory { + @Volatile private var counter = 0L + + val bytesRead: Long + get() = counter + + fun reset() { + counter = 0L + } + + override fun createDataSource(): DataSource = Counting(delegate.createDataSource()) + + private inner class Counting(private val inner: DataSource) : DataSource { + override fun open(dataSpec: DataSpec): Long = inner.open(dataSpec) + + override fun read(buffer: ByteArray, offset: Int, length: Int): Int { + val read = inner.read(buffer, offset, length) + if (read > 0) counter += read + return read + } + + override fun addTransferListener(transferListener: TransferListener) = inner.addTransferListener(transferListener) + + override fun getUri(): Uri? = inner.uri + + override fun getResponseHeaders(): Map> = inner.responseHeaders + + override fun close() = inner.close() + } + } +} diff --git a/android/app/src/main/cpp/media3_ffmpeg_demuxer/ffmpeg_demuxer_jni.cc b/android/app/src/main/cpp/media3_ffmpeg_demuxer/ffmpeg_demuxer_jni.cc index eac4b4e56..3519cc1f7 100644 --- a/android/app/src/main/cpp/media3_ffmpeg_demuxer/ffmpeg_demuxer_jni.cc +++ b/android/app/src/main/cpp/media3_ffmpeg_demuxer/ffmpeg_demuxer_jni.cc @@ -5,30 +5,33 @@ * is served by a media3 ExtractorInput, so FFmpeg parses the container while * media3 keeps owning decoders, renderers, and the whole playback surface. * - * The hard seam is seeking. media3's Extractor is a pull API: an extractor + * The seam used to be seeking. media3's Extractor is a pull API: an extractor * cannot move the input; it returns RESULT_SEEK and the loader re-opens the - * data source at the requested position. libavformat expects a random-access - * AVIO layer and seeks whenever it pleases (MP4 moov at end of file, AVI idx1, - * MKV Cues, avformat_find_stream_info read-ahead). The bridge reconciles the - * two with a deferral protocol: + * data source. libavformat expects random access and seeks whenever it pleases + * (MP4 moov at end of file, AVI idx1, MKV Cues, MPEG-PS binary search), so + * serving it from the loader alone meant unwinding the in-flight libavformat + * call on every backward jump and replaying the whole open — which a + * one-shot lazy parse like matroska's Cues cannot survive (#2096). * - * - The read callback only serves when the AVIO logical position equals the - * ExtractorInput position. Any divergence aborts the in-flight libavformat - * call with AVERROR_NEED_SEEK and records the wanted position. - * - Kotlin maps that to RESULT_SEEK (or, when the loader already moved the - * input to exactly that position, to a local resume). After the input is in - * place, resumeAfterSeek() runs avio_seek(), whose success invalidates the - * stale AVIO buffers, and the interrupted operation is retried. - * - During avformat_open_input/avformat_find_stream_info a block cache serves - * header re-reads from memory so each distinct seek position costs at most - * one loader round trip even though open restarts from zero. + * libavformat now gets what it actually requires: the Kotlin side hands this + * shim a random-access byte source (FfmpegRandomAccessSource), so every AVIO + * read succeeds at the position libavformat asked for and nothing is ever + * unwound. Two paths serve reads, chosen per read: + * + * - The loader's ExtractorInput when it already stands at that position. This + * is the steady state for sample delivery, and it keeps media3's byte + * accounting, back-pressure and load-error policy in charge of streaming. + * - The random-access source otherwise (header, index, post-seek jumps). + * + * Kotlin nudges the loader back into step after a jump by returning + * RESULT_SEEK to nativeLogicalPosition(); that is an optimization, never a + * correctness requirement, so no budgets or retries guard it. */ #include #include #include #include -#include #include extern "C" { @@ -56,18 +59,12 @@ extern "C" { namespace { -// Returned by the AVIO callbacks to unwind into Java when the ExtractorInput -// cannot serve the requested position synchronously. FFERRTAG values are -// distinct from every real AVERROR code. -const int AVERROR_NEED_SEEK = FFERRTAG('P', 'L', 'Z', 'S'); - // readPacket() result codes written to out[0]. Negative values are raw -// AVERROR codes; ERR_JAVA signals the ExtractorInput threw and Kotlin must +// AVERROR codes; ERR_JAVA signals the byte source threw and Kotlin must // rethrow the stored IOException. const jint CODE_PACKET = 0; const jint CODE_EOF = 1; -const jint CODE_NEED_SEEK = 2; -const jint CODE_GROW = 3; +const jint CODE_GROW = 2; const jint ERR_NOT_OPEN = -101; const jint ERR_JAVA = -102; @@ -108,21 +105,14 @@ const jint ROLE_FLAG_MAIN = 1; const jint ROLE_FLAG_DESCRIPTION = 64; // C.ROLE_FLAG_DESCRIPTION const jint ROLE_FLAG_HARD_OF_HEARING = 512; // C.ROLE_FLAG_HARD_OF_HEARING -constexpr int64_t kNoPendingSeek = -1; -constexpr int64_t kNoPendingTimeSeek = INT64_MIN; -constexpr size_t kMaxCacheBytes = 24u * 1024 * 1024; constexpr jint kReadChunkBytes = 64 * 1024; -struct CacheBlock { - int64_t pos; - std::vector data; -}; - struct DemuxState { JavaVM* vm = nullptr; jobject input = nullptr; jmethodID midPosition = nullptr; jmethodID midRead = nullptr; + jmethodID midReadAt = nullptr; jmethodID midLength = nullptr; jbyteArray readBuffer = nullptr; @@ -130,25 +120,14 @@ struct DemuxState { AVIOContext* avio = nullptr; bool opened = false; + // Byte position libavformat will read next. Authoritative: the AVIO seek + // callback moves it, and both read paths serve exactly this offset. int64_t logicalPos = 0; - int64_t pendingSeek = kNoPendingSeek; - // Presentation-time seek requested via Extractor.seek(); executed by the - // next nativeReadPacket, where the input proxy is bound and deferrals can - // round-trip through the loader. - int64_t pendingTimeSeekUs = kNoPendingTimeSeek; - // Loader round trips consumed by the pending time seek; a seek whose - // avformat_seek_file keeps deferring past this budget is abandoned so - // playback continues instead of ping-ponging forever. - int seekAttempts = 0; // format->start_time in microseconds; subtracted from delivered pts so // media3's zero-based timeline matches the duration we report (MPEG-PS/TS // containers commonly start at a nonzero PCR). int64_t startTimeUs = 0; - bool cacheActive = true; - std::deque cache; - size_t cacheBytes = 0; - AVPacket* packet = nullptr; AVPacket* filteredPacket = nullptr; // A packet that outgrew the Java-side buffer: already read and filtered, @@ -208,28 +187,16 @@ jint callRead(JNIEnv* env, DemuxState* s, jint length) { return n; } -void cacheClear(DemuxState* s) { - s->cache.clear(); - s->cacheBytes = 0; -} - -void cacheAppend(DemuxState* s, int64_t pos, const uint8_t* data, size_t size) { - if (!s->cacheActive) return; - s->cache.push_back(CacheBlock{pos, std::vector(data, data + size)}); - s->cacheBytes += size; - while (s->cacheBytes > kMaxCacheBytes && !s->cache.empty()) { - s->cacheBytes -= s->cache.front().data.size(); - s->cache.pop_front(); +// Same contract as callRead, but at an absolute position: the Kotlin source +// reopens its handle when the position is not where it stands. +jint callReadAt(JNIEnv* env, DemuxState* s, int64_t position, jint length) { + jint n = env->CallIntMethod(s->input, s->midReadAt, static_cast(position), s->readBuffer, length); + if (javaPending(env)) { + javaClear(env); + s->javaError = true; + return -1; } -} - -const CacheBlock* cacheFind(const DemuxState* s, int64_t pos) { - for (const CacheBlock& block : s->cache) { - if (pos >= block.pos && pos < block.pos + static_cast(block.data.size())) { - return █ - } - } - return nullptr; + return n; } int avioReadPacket(void* opaque, uint8_t* buf, int bufSize) { @@ -238,55 +205,22 @@ int avioReadPacket(void* opaque, uint8_t* buf, int bufSize) { if (env == nullptr) return AVERROR(EIO); if (bufSize <= 0) return 0; - // Cache-first: bytes already read are served at any logical position, - // which lets restarted header phases (probe → hdrl → idx1 at end of file - // → back to movi) replay through seeks that never touch the loader. - const CacheBlock* block = cacheFind(s, s->logicalPos); - if (block != nullptr) { - size_t offset = static_cast(s->logicalPos - block->pos); - size_t n = std::min(bufSize, block->data.size() - offset); - std::memcpy(buf, block->data.data() + offset, n); - s->logicalPos += static_cast(n); - return static_cast(n); - } - - // A read at or beyond the known end of file is EOF regardless of where the - // loader sits. EOF yields no bytes, so it can never enter the block cache; - // deferring it would let an open that alternates between low offsets and an - // end-of-file probe (avi idx1/ODML index scans) ping-pong against the - // loader forever, one round trip per attempt. - int64_t knownLength = callLength(env, s); + const jint want = bufSize < kReadChunkBytes ? bufSize : kReadChunkBytes; + // Prefer the loader's input when it already stands here: those bytes then + // count as media3's own load progress, which is what drives its buffering + // and back-pressure. Anywhere else, read at the absolute position. + const int64_t inputPos = callPosition(env, s); if (javaPending(env)) { javaClear(env); s->javaError = true; return AVERROR(EIO); } - if (knownLength >= 0 && s->logicalPos >= knownLength) return AVERROR_EOF; - - // Uncached read: the input must be exactly where libavformat thinks it is. - // Any divergence aborts the in-flight call so Kotlin reconciles through - // avio_seek, which also invalidates the stale AVIO buffers. - if (s->logicalPos != callPosition(env, s)) { - if (javaPending(env)) { - javaClear(env); - s->javaError = true; - return AVERROR(EIO); - } - LOGW( - "read defer: logical=%lld input=%lld active=%d cacheBlocks=%zu cacheBytes=%zu first=%lld last=%lld", - (long long)s->logicalPos, (long long)callPosition(env, s), s->cacheActive ? 1 : 0, s->cache.size(), - s->cacheBytes, s->cache.empty() ? -1 : (long long)s->cache.front().pos, - s->cache.empty() ? -1 : (long long)s->cache.back().pos); - s->pendingSeek = s->logicalPos; - return AVERROR_NEED_SEEK; - } - - jint n = callRead(env, s, bufSize < kReadChunkBytes ? bufSize : kReadChunkBytes); + const jint n = inputPos == s->logicalPos ? callRead(env, s, want) : callReadAt(env, s, s->logicalPos, want); if (s->javaError) return AVERROR(EIO); - if (n <= 0) return AVERROR_EOF; + if (n < 0) return AVERROR(EIO); + if (n == 0) return AVERROR_EOF; env->GetByteArrayRegion(s->readBuffer, 0, n, reinterpret_cast(buf)); - cacheAppend(s, s->logicalPos, buf, static_cast(n)); s->logicalPos += n; return n; } @@ -323,21 +257,23 @@ int64_t avioSeek(void* opaque, int64_t offset, int whence) { target = offset; } - // Seeks are pure bookkeeping and never touch the loader. Restarted header - // replays hop through cached regions for free; a seek into an uncached - // region surfaces as a deferral on the next uncached read, with the input - // still where the loader left it. avio invalidates its buffers after this - // successful callback. - // - // Deliberately does NOT touch s->pendingSeek: an unconsumed deferral target - // must survive the avformat unwind until Kotlin consumes it. + // Pure bookkeeping: the next read serves this position from whichever path + // can, so a seek can no longer fail or need reconciling. avio invalidates + // its own buffers after this successful callback. s->logicalPos = target; - // A deferral leaves AVIOContext::error sticky; once a seek has been adopted - // it must not replay into later reads. - if (s->avio != nullptr) s->avio->error = 0; return target; } +// Probing (ffio_ensure_seekback) can swap the AVIO buffer for a larger +// allocation and free the original, so the pointer handed to +// avio_alloc_context may be stale. Free whichever buffer the context +// currently holds, exactly once. +void freeAvio(DemuxState* s) { + if (s->avio == nullptr) return; + av_freep(&s->avio->buffer); + avio_context_free(&s->avio); +} + void closeState() { DemuxState* s = gState; if (s == nullptr) return; // Close without (or after) an open is a no-op. @@ -357,14 +293,7 @@ void closeState() { if (s->format != nullptr) { avformat_close_input(&s->format); } - if (s->avio != nullptr) { - // Probing (ffio_ensure_seekback) can swap the AVIO buffer for a larger - // allocation and free the original, so the pointer handed to - // avio_alloc_context may be stale. Free whichever buffer the context - // currently holds, exactly once. - av_freep(&s->avio->buffer); - avio_context_free(&s->avio); - } + freeAvio(s); if (env != nullptr && s->input != nullptr) { if (s->readBuffer != nullptr) env->DeleteGlobalRef(s->readBuffer); env->DeleteGlobalRef(s->input); @@ -373,8 +302,6 @@ void closeState() { gState = nullptr; } -bool deferred(DemuxState* s, int err) { return err == AVERROR_NEED_SEEK || s->pendingSeek != kNoPendingSeek; } - const char* videoMime(AVCodecID id) { switch (id) { case AV_CODEC_ID_H264: @@ -528,10 +455,9 @@ Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeProbeFormat(JNIEnv* env, return env->NewStringUTF(format->name); } -// Returns 0 on success, CODE_NEED_SEEK when a loader round trip is required, -// or a negative AVERROR. The stream count is NOT the return value: positive -// counts share the result-code namespace (CODE_NEED_SEEK == 2 would collide -// with every two-stream container), so callers use nativeStreamCount(). +// Returns 0 on success or a negative AVERROR (ERR_JAVA when the byte source +// itself threw). The stream count is NOT the return value: positive counts +// would share the result-code namespace, so callers use nativeStreamCount(). JNIEXPORT jint JNICALL Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeOpen(JNIEnv* env, jobject thiz, jobject input) { StateLock lock; @@ -544,7 +470,6 @@ Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeOpen(JNIEnv* env, jobjec gState->vm = gVm; } DemuxState* s = gState; - if (s->avio != nullptr) s->avio->error = 0; // FindClass on the declared interface (not GetObjectClass on the concrete // proxy): JNI dispatches interface method IDs virtually, and the named @@ -556,9 +481,10 @@ Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeOpen(JNIEnv* env, jobjec } jmethodID midPosition = env->GetMethodID(inputClass, "position", "()J"); jmethodID midRead = env->GetMethodID(inputClass, "read", "([BI)I"); + jmethodID midReadAt = env->GetMethodID(inputClass, "readAt", "(J[BI)I"); jmethodID midLength = env->GetMethodID(inputClass, "length", "()J"); env->DeleteLocalRef(inputClass); - if (midPosition == nullptr || midRead == nullptr || midLength == nullptr) { + if (midPosition == nullptr || midRead == nullptr || midReadAt == nullptr || midLength == nullptr) { javaClear(env); return ERR_NOT_OPEN; } @@ -573,6 +499,7 @@ Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeOpen(JNIEnv* env, jobjec s->input = env->NewGlobalRef(input); s->midPosition = midPosition; s->midRead = midRead; + s->midReadAt = midReadAt; s->midLength = midLength; jbyteArray readBuffer = env->NewByteArray(kReadChunkBytes); s->readBuffer = static_cast(env->NewGlobalRef(readBuffer)); @@ -584,44 +511,29 @@ Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeOpen(JNIEnv* env, jobjec if (s->packet == nullptr || s->filteredPacket == nullptr) return AVERROR(ENOMEM); } + // One AVIOContext per item: a stale buffer or a sticky error from the + // previous item must not leak into this open, and reallocating is cheaper + // than reasoning about which of libavformat's internal fields to reset. + freeAvio(s); + constexpr int kAvioBufferSize = 64 * 1024; + unsigned char* avioBuffer = static_cast(av_malloc(kAvioBufferSize)); + if (avioBuffer == nullptr) return AVERROR(ENOMEM); + s->avio = avio_alloc_context( + avioBuffer, kAvioBufferSize, /*write_flag=*/0, s, avioReadPacket, + /*write_packet=*/nullptr, avioSeek); if (s->avio == nullptr) { - constexpr int kAvioBufferSize = 64 * 1024; - unsigned char* avioBuffer = static_cast(av_malloc(kAvioBufferSize)); - s->avio = avio_alloc_context( - avioBuffer, kAvioBufferSize, /*write_flag=*/0, s, avioReadPacket, - /*write_packet=*/nullptr, avioSeek); - if (s->avio == nullptr) return AVERROR(ENOMEM); + av_freep(&avioBuffer); + return AVERROR(ENOMEM); } s->javaError = false; - s->pendingSeek = kNoPendingSeek; - s->pendingTimeSeekUs = kNoPendingTimeSeek; - s->seekAttempts = 0; s->packetPending = false; - // Every open attempt restarts from scratch and replays the header phase - // through the block cache. Resuming avformat_find_stream_info across - // synthetic IO aborts is not something libavformat contracts to survive - // (observed as a NULL deref inside avformat_find_stream_info), and with the - // cache a full replay costs nothing on the wire. - // - // The cache must survive retries within one open: matroska's header phase - // jumps backward twice (SeekHead -> Cues -> clusters), and clearing here — - // per attempt — makes every retry replay against an empty cache and - // ping-pong against the loader until the reconcile cap kills the open. - // Invalidation is per media item and lives in nativeResetCache, called - // from FfmpegExtractor.init when media3 binds a new source. + s->logicalPos = 0; + // Nothing is ever resumed across calls now: every read succeeds at the + // position libavformat asked for, so open_input and find_stream_info each + // run once, to completion. avformat_close_input(&s->format); s->opened = false; - s->cacheActive = true; - - int64_t seekResult = avio_seek(s->avio, 0, SEEK_SET); - if (s->pendingSeek != kNoPendingSeek) { - return CODE_NEED_SEEK; - } - if (seekResult < 0) { - LOGW("open seek0 failed=%lld sticky=%d", (long long)seekResult, s->avio != nullptr ? s->avio->error : 0); - return static_cast(seekResult); - } s->format = avformat_alloc_context(); if (s->format == nullptr) return AVERROR(ENOMEM); @@ -631,33 +543,22 @@ Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeOpen(JNIEnv* env, jobjec s->format->flags |= AVFMT_FLAG_GENPTS; int err = avformat_open_input(&s->format, "", nullptr, nullptr); - if (deferred(s, err)) { - LOGW("open_input defer pending=%lld err=%d", (long long)s->pendingSeek, err); - return CODE_NEED_SEEK; - } + if (s->javaError) return ERR_JAVA; if (err < 0) { LOGW("avformat_open_input failed: %d", err); return err; } err = avformat_find_stream_info(s->format, nullptr); - if (deferred(s, err)) { - LOGW("find_info defer pending=%lld err=%d", (long long)s->pendingSeek, err); - return CODE_NEED_SEEK; - } + if (s->javaError) return ERR_JAVA; if (err < 0) { LOGW("avformat_find_stream_info failed: %d", err); return err; } - s->opened = true; // Zero-base the timeline: delivered pts subtract this so media3 sees // [0, duration] even when the container starts at a nonzero PCR/PTS. s->startTimeUs = s->format->start_time != AV_NOPTS_VALUE ? s->format->start_time : 0; - - // Header work is done: packet reads are sequential and every backward move - // goes through a real loader seek. - s->cacheActive = false; - cacheClear(s); + s->opened = true; bool anySupported = false; for (unsigned i = 0; i < s->format->nb_streams; i++) { @@ -733,18 +634,6 @@ JNIEXPORT void JNICALL Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeC closeState(); } -JNIEXPORT void JNICALL Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeResetCache(JNIEnv* env, jobject thiz) { - StateLock lock; - // Per-item boundary: media3 reuses extractor instances across sources, and - // a new item's header occupies the same low offsets as the previous item's. - // Serving those stale bytes would corrupt the new open, so drop everything; - // nativeOpen re-arms the active flag for its retry phase. - DemuxState* s = gState; - if (s == nullptr) return; - cacheClear(s); - s->cacheActive = true; -} - JNIEXPORT jint JNICALL Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeStreamCount(JNIEnv* env, jobject thiz) { StateLock lock; if (gState == nullptr || gState->format == nullptr) return ERR_NOT_OPEN; @@ -932,7 +821,6 @@ Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeAttachmentData(JNIEnv* e // [1]=streamIndex [2]=ptsUs (start-time normalized; packets without any // timestamp are dropped, media3 cannot schedule them) [3]=flags [4]=size // [5]=packet position in input coordinates [6]=durationUs (-1 unknown). -// Deferral targets travel through nativeConsumePendingSeek, not this array. JNIEXPORT jint JNICALL Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeReadPacket( JNIEnv* env, jobject thiz, jbyteArray buffer, jlongArray out) { StateLock lock; @@ -942,94 +830,7 @@ JNIEXPORT jint JNICALL Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeR jint capacity = env->GetArrayLength(buffer); jlong result[7] = {}; s->javaError = false; - s->pendingSeek = kNoPendingSeek; - - int stickyRecovers = 0; while (true) { - // A presentation-time seek from Extractor.seek() runs here, where the - // input proxy is bound and a deferral can round-trip through the loader. - // The demuxer picks the byte position and resets its own parse state — - // jumping the AVIO position under it would leave demuxer-private state - // (avi->remaining, PES continuations) pointing into the old stream. - // avformat_seek_file restarts from scratch on every retry; the block - // cache is re-armed so index/probe reads replayed across loader round - // trips (matroska Cues, mpegps binary search) stay free on the wire. - if (s->pendingTimeSeekUs != kNoPendingTimeSeek) { - if (++s->seekAttempts > 64) { - // Non-converging seek (each attempt = one loader round trip): - // abandon it and keep delivering from the current position rather - // than ping-ponging forever. - LOGW( - "time seek to %lld abandoned after %d loader round trips", (long long)s->pendingTimeSeekUs, - s->seekAttempts - 1); - s->pendingTimeSeekUs = kNoPendingTimeSeek; - s->cacheActive = false; - cacheClear(s); - } else { - if (s->bsf != nullptr) { - for (unsigned i = 0; i < s->nbStreams; i++) { - if (s->bsf[i] != nullptr) av_bsf_flush(s->bsf[i]); - } - } - s->packetPending = false; - s->cacheActive = true; - int64_t ts = s->pendingTimeSeekUs + s->startTimeUs; - int err = avformat_seek_file(s->format, -1, INT64_MIN, ts, ts, 0); - LOGD( - "time seek to %lld: err=%d logical=%lld attempts=%d", (long long)ts, err, (long long)s->logicalPos, - s->seekAttempts); - if (deferred(s, err)) { - if (s->pendingSeek == kNoPendingSeek) { - if (++stickyRecovers > 16 || s->avio == nullptr) return ERR_JAVA; - s->avio->error = 0; - continue; - } - // pendingTimeSeekUs stays set: the retry after the loader round - // trip re-runs the whole seek. - result[0] = CODE_NEED_SEEK; - env->SetLongArrayRegion(out, 0, 7, result); - return CODE_NEED_SEEK; - } - if (s->javaError) { - result[0] = ERR_JAVA; - env->SetLongArrayRegion(out, 0, 7, result); - return ERR_JAVA; - } - s->pendingTimeSeekUs = kNoPendingTimeSeek; - if (err < 0) { - // Unseekable stream or demuxer refusal: keep playing from the - // current position instead of killing the session; media3 will - // decode-discard toward the target if it can. - LOGW("avformat_seek_file to %lld failed: %d", (long long)ts, err); - } else { - // Align the loader with the demuxer's chosen byte position BEFORE - // the first read. avformat_seek_file resolves in memory (cues) and - // typically leaves the AVIO position behind the loader's SeekMap - // guess; letting av_read_frame discover that and unwind with a - // synthetic IO error poisons matroskadec into resync, which skips - // the keyframe cluster — the video renderer then starves until the - // next keyframe while audio keeps advancing, and the resume-stall - // watchdog seeks in a loop. The cache stays armed across this round - // trip (until the first packet delivers) so replayed header bytes - // stay free. - int64_t inputPos = callPosition(env, s); - if (javaPending(env)) { - javaClear(env); - s->javaError = true; - result[0] = ERR_JAVA; - env->SetLongArrayRegion(out, 0, 7, result); - return ERR_JAVA; - } - if (inputPos >= 0 && inputPos != s->logicalPos) { - s->pendingSeek = s->logicalPos; - result[0] = CODE_NEED_SEEK; - env->SetLongArrayRegion(out, 0, 7, result); - return CODE_NEED_SEEK; - } - } - } - } - if (s->packetPending) { // Oversized packet held across a CODE_GROW round trip: deliver it now // instead of reading (and losing) a new frame. @@ -1037,19 +838,6 @@ JNIEXPORT jint JNICALL Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeR } else { av_packet_unref(s->packet); int err = av_read_frame(s->format, s->packet); - if (deferred(s, err)) { - if (s->pendingSeek == kNoPendingSeek) { - // Sticky AVIO error replayed from a deferral Kotlin already - // reconciled: clear it and retry instead of surfacing a targetless - // seek. - if (++stickyRecovers > 16 || s->avio == nullptr) return ERR_JAVA; - s->avio->error = 0; - continue; - } - result[0] = CODE_NEED_SEEK; - env->SetLongArrayRegion(out, 0, 7, result); - return CODE_NEED_SEEK; - } if (err == AVERROR_EOF) { result[0] = CODE_EOF; env->SetLongArrayRegion(out, 0, 7, result); @@ -1111,13 +899,6 @@ JNIEXPORT jint JNICALL Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeR } ptsUs -= s->startTimeUs; - // Post-seek convergence is over once a packet flows; steady-state packet - // reads are sequential, so drop the replay cache until the next seek. - if (s->cacheActive) { - s->cacheActive = false; - cacheClear(s); - } - result[0] = CODE_PACKET; result[1] = s->packet->stream_index; result[2] = ptsUs; @@ -1132,42 +913,40 @@ JNIEXPORT jint JNICALL Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeR } JNIEXPORT jlong JNICALL -Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeConsumePendingSeek(JNIEnv* env, jobject thiz) { +Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeLogicalPosition(JNIEnv* env, jobject thiz) { StateLock lock; - if (gState == nullptr) return -9223372036854775807L - 1; // Long.MIN_VALUE - int64_t pending = gState->pendingSeek; - gState->pendingSeek = kNoPendingSeek; - return pending; + DemuxState* s = gState; + if (s == nullptr || !s->opened) return -1; + return static_cast(s->logicalPos); } -// Re-establishes the AVIO position after the loader moved the input to -// `position`. Must only be called when the ExtractorInput is already there. +// Seeks to a presentation time (media3 timeline microseconds, start-time +// normalized). Called on the loader thread, where blocking IO belongs: +// libavformat reads its own index through the random-access source, so this +// either lands on the target keyframe or reports that the container has no +// usable index. JNIEXPORT jint JNICALL -Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeResumeAfterSeek(JNIEnv* env, jobject thiz, jlong position) { +Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeSeek(JNIEnv* env, jobject thiz, jlong timeUs) { StateLock lock; DemuxState* s = gState; - if (s == nullptr || s->avio == nullptr) return ERR_NOT_OPEN; - s->pendingSeek = kNoPendingSeek; + if (s == nullptr || !s->opened || s->format == nullptr) return ERR_NOT_OPEN; s->javaError = false; - int64_t result = avio_seek(s->avio, position, SEEK_SET); - if (result >= 0) return 0; - if (s->pendingSeek != kNoPendingSeek) return CODE_NEED_SEEK; - return static_cast(result); -} - -// Records a presentation-time seek (media3 timeline microseconds, i.e. -// start-time normalized). Executed by the next nativeReadPacket, which owns -// a bound input proxy and can defer through the loader; running -// avformat_seek_file here would read against a dead input. -JNIEXPORT void JNICALL -Java_com_edde746_plezy_exoplayer_FfmpegDemuxerJni_nativeSeekTo(JNIEnv* env, jobject thiz, jlong timeUs) { - StateLock lock; - DemuxState* s = gState; - if (s == nullptr || !s->opened) return; - s->pendingTimeSeekUs = timeUs; - s->seekAttempts = 0; // Any packet parked for a CODE_GROW retry belongs to the pre-seek stream. s->packetPending = false; + if (s->bsf != nullptr) { + for (unsigned i = 0; i < s->nbStreams; i++) { + if (s->bsf[i] != nullptr) av_bsf_flush(s->bsf[i]); + } + } + const int64_t ts = timeUs + s->startTimeUs; + const int err = avformat_seek_file(s->format, -1, INT64_MIN, ts, ts, 0); + if (s->javaError) return ERR_JAVA; + if (err < 0) { + LOGW("seek to %lld failed: %d", (long long)ts, err); + return err; + } + LOGD("seek to %lld: logical=%lld", (long long)ts, (long long)s->logicalPos); + return 0; } } // extern "C" diff --git a/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/ExoPlayerCore.kt b/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/ExoPlayerCore.kt index d2ff045cf..f25b2ecd8 100644 --- a/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/ExoPlayerCore.kt +++ b/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/ExoPlayerCore.kt @@ -746,6 +746,11 @@ class ExoPlayerCore(private val activity: Activity) : } httpDataSourceFactory = httpFactory dataSourceFactory = DefaultDataSource.Factory(activity, httpFactory) + // libavformat owns its own seeking, so the demuxer reads through a + // random-access source (FfmpegRandomAccessSource). Recording the DataSpec + // media3 opens lets it clone that spec — same URI, cache key, flags and + // request headers — instead of re-deriving one. + val recordingFactory = RecordingDataSourceFactory(dataSourceFactory!!) val extractorsFactory = DefaultExtractorsFactory() // High-bitrate Plex DVR MPEG-TS recordings can have sparse PCR packets; the default // 600-packet window may leave duration unknown and seeking disabled. @@ -773,55 +778,73 @@ class ExoPlayerCore(private val activity: Activity) : // Reads this.dvMode and this.demuxerPreference each time (not captured) so // DV7→8.1 retry and demuxer-mode changes can reload without // reinitializing the player. - val wrappedExtractorsFactory = androidx.media3.extractor.ExtractorsFactory { - val currentDvMode = this.dvMode - val doviEnabled = currentDvMode != DvConversionMode.DISABLED - val currentDemuxerPreference = this.demuxerPreference - Log.i(TAG, "[init] extractors: demuxer=${currentDemuxerPreference.wireName}") - // media3 invokes the factory once per player session and reuses the - // extractor instances for every item, so the sniff reads the live - // preference instead of a captured one. - val liveDemuxerPreference = { this@ExoPlayerCore.demuxerPreference } - val ffmpegFirst = - listOfNotNull( - FfmpegExtractor.create( - liveDemuxerPreference, - currentDvMode, - subtitleParserFactory, - handler - )?.also { activeFfmpegExtractor = it } + val wrappedExtractorsFactory = object : androidx.media3.extractor.ExtractorsFactory { + // media3's progressive path always supplies the item URI. Without one + // the demuxer cannot be handed random access, so FFmpeg stays out + // rather than failing later on its first index jump. + override fun createExtractors(): Array { + Log.w(TAG, "[init] extractors: no item uri, ffmpeg demuxer unavailable") + return buildExtractors(null) + } + + override fun createExtractors( + uri: Uri, + responseHeaders: Map> + ): Array = buildExtractors(uri) + + private fun buildExtractors(uri: Uri?): Array { + val currentDvMode = this@ExoPlayerCore.dvMode + val doviEnabled = currentDvMode != DvConversionMode.DISABLED + val currentDemuxerPreference = this@ExoPlayerCore.demuxerPreference + Log.i(TAG, "[init] extractors: demuxer=${currentDemuxerPreference.wireName}") + // media3 invokes the factory once per item and reuses the extractor + // instances, so the sniff reads the live preference instead of a + // captured one. + val liveDemuxerPreference = { this@ExoPlayerCore.demuxerPreference } + val ffmpegFirst = listOfNotNull( + uri?.let { itemUri -> + val io = FfmpegRandomAccessSource(dataSourceFactory!!) { recordingFactory.specFor(itemUri) } + FfmpegExtractor.create( + liveDemuxerPreference, + currentDvMode, + subtitleParserFactory, + handler, + io + )?.also { activeFfmpegExtractor = it } + } ) - ( - ffmpegFirst + extractorsFactory.createExtractors().map { extractor -> - when { - extractor is MatroskaExtractor -> { - val withTextPipeline = OutputWrappingExtractor(extractor) { out -> - AssSubtitleExtractorOutput(SubtitleTranscodingExtractorOutput(out, subtitleParserFactory), handler) + return ( + ffmpegFirst + extractorsFactory.createExtractors().map { extractor -> + when { + extractor is MatroskaExtractor -> { + val withTextPipeline = OutputWrappingExtractor(extractor) { out -> + AssSubtitleExtractorOutput(SubtitleTranscodingExtractorOutput(out, subtitleParserFactory), handler) + } + if (doviEnabled) { + DoviExtractorWrapper(withTextPipeline, currentDvMode) { level, prefix, message -> + emitLog(level, prefix, message) + }.also { + activeDoviMkvWrapper = it + } + } else { + withTextPipeline + } } - if (doviEnabled) { - DoviExtractorWrapper(withTextPipeline, currentDvMode) { level, prefix, message -> + doviEnabled && (extractor is Mp4Extractor || extractor is FragmentedMp4Extractor) -> { + DoviExtractorWrapper(extractor, currentDvMode) { level, prefix, message -> emitLog(level, prefix, message) }.also { - activeDoviMkvWrapper = it + activeDoviMp4Wrapper = it } - } else { - withTextPipeline } + else -> extractor } - doviEnabled && (extractor is Mp4Extractor || extractor is FragmentedMp4Extractor) -> { - DoviExtractorWrapper(extractor, currentDvMode) { level, prefix, message -> - emitLog(level, prefix, message) - }.also { - activeDoviMp4Wrapper = it - } - } - else -> extractor } - } - ).toTypedArray() + ).toTypedArray() + } } - val mediaSourceFactory = DefaultMediaSourceFactory(dataSourceFactory!!, wrappedExtractorsFactory) + val mediaSourceFactory = DefaultMediaSourceFactory(recordingFactory, wrappedExtractorsFactory) .setSubtitleParserFactory(subtitleParserFactory) playbackMediaSourceFactory = mediaSourceFactory diff --git a/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/FfmpegDemuxerJni.kt b/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/FfmpegDemuxerJni.kt index 86423bf0b..db63ad9bf 100644 --- a/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/FfmpegDemuxerJni.kt +++ b/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/FfmpegDemuxerJni.kt @@ -15,25 +15,32 @@ import androidx.media3.common.util.UnstableApi */ @OptIn(UnstableApi::class) internal object FfmpegDemuxerJni { - /** Re-entry point for the AVIO callbacks while a native call is in flight. */ + /** Byte source the AVIO callbacks read through while a native call is in flight. */ interface Input { - /** Current byte position of the underlying [androidx.media3.extractor.ExtractorInput]. */ + /** Current byte position of the loader's [androidx.media3.extractor.ExtractorInput]. */ fun position(): Long /** - * Reads up to [length] bytes into [buf]. Returns the number of bytes read, - * `0` at end of input, or `-1` after storing the failure in [lastError]. + * Reads up to [length] bytes from the loader's input at its current + * position. Returns the number of bytes read, `0` at end of input, or `-1` + * after storing the failure in [lastError]. */ fun read(buf: ByteArray, length: Int): Int + /** + * Reads up to [length] bytes at absolute [position], wherever the loader + * stands. Same return contract as [read]. This is what lets libavformat + * seek its own index without the shim unwinding the in-flight call. + */ + fun readAt(position: Long, buf: ByteArray, length: Int): Int + /** Total input length in bytes, or `-1` when unknown. */ fun length(): Long var lastError: String? } - // readPacket out[] indices. Deferral targets travel through - // [nativeConsumePendingSeek], not this array. + // readPacket out[] indices. const val OUT_CODE = 0 const val OUT_STREAM_INDEX = 1 const val OUT_PTS_US = 2 @@ -46,8 +53,7 @@ internal object FfmpegDemuxerJni { // Result codes written to out[OUT_CODE]; negative values are raw AVERRORs. const val CODE_PACKET = 0 const val CODE_EOF = 1 - const val CODE_NEED_SEEK = 2 - const val CODE_GROW = 3 + const val CODE_GROW = 2 const val ERR_JAVA = -102 // streamInfo long[] layout. @@ -81,10 +87,10 @@ internal object FfmpegDemuxerJni { external fun nativeProbeFormat(header: ByteArray): String? /** - * Opens (and resumes opening) the demuxer against [input]. Returns 0 on - * success, [CODE_NEED_SEEK] when the loader must move the input first, or a - * negative AVERROR. The stream count is read separately via - * [nativeStreamCount]; it must not ride the same namespace as result codes. + * Opens the demuxer against [input]. Returns 0 on success or a negative + * AVERROR ([ERR_JAVA] when the input itself failed). The stream count is read + * separately via [nativeStreamCount]; it must not ride the same namespace as + * result codes. */ external fun nativeOpen(input: Input): Int @@ -120,19 +126,14 @@ internal object FfmpegDemuxerJni { */ external fun nativeReadPacket(buffer: ByteArray, out: LongArray): Int - /** Returns the recorded seek target, or [Long.MIN_VALUE] when none. */ - external fun nativeConsumePendingSeek(): Long - - /** Adopts [position] as the AVIO position; the input must already be there. */ - external fun nativeResumeAfterSeek(position: Long): Int + /** Byte position libavformat will read next, or -1 when not open. */ + external fun nativeLogicalPosition(): Long /** - * Records a presentation-time seek (media3 timeline microseconds). The next - * [nativeReadPacket] executes it via avformat_seek_file, deferring through - * the loader as needed; the demuxer picks the byte position itself. + * Seeks to [timeUs] (media3 timeline microseconds, i.e. start-time + * normalized). The demuxer resolves the byte position itself through its own + * index. Returns 0 on success, or a negative AVERROR when the container has + * no usable index and libavformat refused the seek. */ - external fun nativeSeekTo(timeUs: Long) - - /** Drops cached header blocks; called when the extractor binds a new source. */ - external fun nativeResetCache() + external fun nativeSeek(timeUs: Long): Int } diff --git a/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/FfmpegExtractor.kt b/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/FfmpegExtractor.kt index f512cb5bc..76e384929 100644 --- a/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/FfmpegExtractor.kt +++ b/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/FfmpegExtractor.kt @@ -36,10 +36,10 @@ import java.io.IOException * composed in [init] then feeds libass dialogue and transcodes cues the same * way the media3 Matroska path does. * - * Seeking protocol: see `ffmpeg_demuxer_jni.cc`. The extractor never moves the - * input itself; position divergences abort the native call, and Kotlin either - * returns RESULT_SEEK or resumes in place when the loader already positioned - * the input at the requested byte. + * Seeking: libavformat resolves seeks itself, reading its own index through + * [FfmpegRandomAccessSource]. media3's byte hints are ignored, nothing is ever + * unwound, and the loader is merely nudged to follow along so that sample + * delivery keeps streaming on its connection (see [alignLoader]). * * The instance sits before media3's extractors and sniffs everything FFmpeg * can probe while [FfmpegDemuxerPolicy] enables it; media3's list behind it @@ -50,28 +50,24 @@ internal class FfmpegExtractor private constructor( private val preferenceSupplier: () -> FfmpegDemuxerPolicy.Preference, private val dvMode: DvConversionMode, private val subtitleParserFactory: SubtitleParser.Factory, - private val assHandler: AssHandler + private val assHandler: AssHandler, + private val io: FfmpegRandomAccessSource ) : Extractor { companion object { private const val TAG = "FfmpegExtractor" private const val SNIFF_BYTES = 64 * 1024 - private const val MAX_RECONCILES = 24 - - // Loader round trips (RESULT_SEEK) allowed for one item's whole open - // phase; matroska needs ~3, avi with an end-of-file index ~4. - private const val MAX_OPEN_LOADER_ROUND_TRIPS = 96 private const val INITIAL_PACKET_BUFFER_BYTES = 2 * 1024 * 1024 - private const val AUDIO_INDEX_INTERVAL_US = 500_000L /** Null when the native library is unavailable. */ fun create( preferenceSupplier: () -> FfmpegDemuxerPolicy.Preference, dvMode: DvConversionMode, subtitleParserFactory: SubtitleParser.Factory, - assHandler: AssHandler + assHandler: AssHandler, + io: FfmpegRandomAccessSource ): FfmpegExtractor? = if (FfmpegDemuxerJni.available) { - FfmpegExtractor(preferenceSupplier, dvMode, subtitleParserFactory, assHandler) + FfmpegExtractor(preferenceSupplier, dvMode, subtitleParserFactory, assHandler, io) } else { null } @@ -94,10 +90,23 @@ internal class FfmpegExtractor private constructor( } } + override fun readAt(position: Long, buf: ByteArray, length: Int): Int { + lastError = null + return try { + io.readAt(position, buf, length) + } catch (e: IOException) { + lastError = e.message ?: "demuxer random access read failed" + -1 + } + } + override fun length(): Long { - val input = currentInput ?: return -1L - val length = input.length - return if (length == C.LENGTH_UNSET.toLong()) -1L else length + val fromLoader = currentInput?.length ?: C.LENGTH_UNSET.toLong() + if (fromLoader != C.LENGTH_UNSET.toLong()) return fromLoader + // Length-less loader open (progressive live stream): fall back to + // whatever a random-access open already resolved. + val resolved = io.length() + return if (resolved == C.LENGTH_UNSET.toLong()) -1L else resolved } override var lastError: String? = null @@ -107,8 +116,6 @@ internal class FfmpegExtractor private constructor( private var doviOutputWrapper: DoviExtractorOutputWrapper? = null private var durationUs = C.TIME_UNSET private var trackOutputs: Array = emptyArray() - private var primaryStreamIndex = -1 - private var primaryStreamIsAudio = false private var packetBytes = ByteArray(INITIAL_PACKET_BUFFER_BYTES) private val packetParsable = ParsableByteArray() private val packetOut = LongArray(FfmpegDemuxerJni.OUT_LENGTH) @@ -122,12 +129,11 @@ internal class FfmpegExtractor private constructor( // the loader thread, read by ExoPlayerCore.getStats on another thread. @Volatile private var bitrateMeters: Array = emptyArray() - // Seek index for the primary stream: (presentation time, input position) of - // video keyframes or periodic audio packets. The SeekMap reads it on the - // playback thread while the loader thread appends, hence the lock. - private val seekIndexTimes = ArrayList() - private val seekIndexPositions = ArrayList() - private val seekIndexLock = Any() + // Set by seek(), applied by the next read(): media3 calls both on the loader + // thread, but only read() is allowed to throw IOException and the demuxer's + // seek reads its index. + private var pendingSeekUs = C.TIME_UNSET + private var deliveredAnyPacket = false private val sniffScratch = ByteArray(SNIFF_BYTES) @@ -167,9 +173,8 @@ internal class FfmpegExtractor private constructor( override fun init(output: ExtractorOutput) { // media3 reuses extractor instances across sources; drop the previous - // item's header cache so this open cannot replay stale bytes. - FfmpegDemuxerJni.nativeResetCache() - openLoaderRoundTrips = 0 + // item's demuxer and byte source so this open starts clean. + releaseDemuxer() // Same composition as the MP4/MKV paths: DoviConvertingTrackOutput // inspects each video track's codecs string and passes non-DV through. val doviWrapped = if (dvMode != DvConversionMode.DISABLED) { @@ -202,8 +207,16 @@ internal class FfmpegExtractor private constructor( override fun read(input: ExtractorInput, seekPosition: PositionHolder): Int { currentInput = input try { - if (!opened && !openStreams(input, seekPosition)) return Extractor.RESULT_SEEK - return readPacket(input, seekPosition) + if (!opened) openStreams() + applyPendingSeek() + alignLoader(input)?.let { position -> + seekPosition.position = position + return Extractor.RESULT_SEEK + } + // Aligned: the loader serves the samples, so nothing needs a second + // handle open until the demuxer jumps again. + if (io.isOpen) io.close() + return readPacket() } catch (e: Throwable) { Log.w(TAG, "read threw at input=${input.position}", e) throw e @@ -213,28 +226,72 @@ internal class FfmpegExtractor private constructor( } /** - * Hands the presentation-time target to the demuxer: the next read runs - * avformat_seek_file, which picks the byte position and resets libavformat's - * parse state (jumping the AVIO position under the demuxer would leave - * demuxer-private state pointing into the old stream). The byte position - * media3 supplies is only its SeekMap guess; the loader converges on the - * demuxer's choice through the normal deferral round trip. The DV - * converter's buffered NAL state resets here as well. + * Byte position the loader should move to, or null when it already stands + * where libavformat reads. + * + * Reads outside the loader's window are served by [io] regardless, so this + * is an optimization, never a correctness requirement: it keeps sample + * delivery on media3's own connection, which is what drives its byte + * accounting, back-pressure (ProgressiveMediaPeriod watches the input + * position) and load-error policy. Nothing here needs a retry budget. + */ + private fun alignLoader(input: ExtractorInput): Long? { + val demuxerPosition = FfmpegDemuxerJni.nativeLogicalPosition() + if (demuxerPosition < 0 || demuxerPosition == input.position) return null + val length = input.length + // Opening a range at or past the end is a failed request, not an + // alignment; let the read report end of input instead. + if (length != C.LENGTH_UNSET.toLong() && demuxerPosition >= length) return null + return demuxerPosition + } + + /** + * Records the target; [read] performs it. media3's byte position is its + * SeekMap guess and is deliberately ignored — libavformat picks the byte + * position from its own index. */ override fun seek(position: Long, timeUs: Long) { - Log.i(TAG, "extractor seek position=$position timeUs=$timeUs opened=$opened") + Log.i(TAG, "extractor seek timeUs=$timeUs opened=$opened") doviOutputWrapper?.resetTracks() // Buffered LOAS bytes belong to the pre-seek stream; the StreamMuxConfig // survives so audio resumes without waiting for the next config. for (latm in latmOutputs) latm.reset() - if (opened) FfmpegDemuxerJni.nativeSeekTo(timeUs) + pendingSeekUs = timeUs + } + + private fun applyPendingSeek() { + val target = pendingSeekUs + if (target == C.TIME_UNSET) return + pendingSeekUs = C.TIME_UNSET + // media3 issues seek(0) before the first read of every item; a demuxer + // that has not delivered a packet yet is already at the start, and + // seeking would throw away find_stream_info's buffered packets and + // re-read the header region. + if (target == 0L && !deliveredAnyPacket) return + val code = FfmpegDemuxerJni.nativeSeek(target) + if (code == FfmpegDemuxerJni.ERR_JAVA) { + throw IOException(inputProxy.lastError ?: "demuxer input failed") + } + if (code != 0) { + // No usable index in this container: libavformat keeps delivering from + // where it stands and the renderers decode-discard toward the target, + // which is what media3's own extractors do for a cueless file. + Log.w(TAG, "demuxer refused seek to ${target}us: $code") + } } override fun release() { + releaseDemuxer() + } + + private fun releaseDemuxer() { if (opened) { opened = false FfmpegDemuxerJni.nativeClose() } + io.close() + pendingSeekUs = C.TIME_UNSET + deliveredAnyPacket = false // media3's extractors may demux a later item while this instance keeps // the previous one's stream layout; a stale measurement must not leak // into stats. @@ -252,49 +309,15 @@ internal class FfmpegExtractor private constructor( return bitrateMeters.getOrNull(index)?.bitrateBps() } - // The in-call `reconciles` counter resets on every read() invocation, so it - // cannot bound RESULT_SEEK ping-pong across loader restarts. This budget - // spans the whole open of one item (reset in init) and turns a - // non-converging header phase into a clean fallback instead of a hang. - private var openLoaderRoundTrips = 0 - - /** Returns false when a loader seek was requested via [seekPosition]. */ - private fun openStreams(input: ExtractorInput, seekPosition: PositionHolder): Boolean { - var reconciles = 0 - while (true) { - when (val code = FfmpegDemuxerJni.nativeOpen(inputProxy)) { - 0 -> { - prepareTracks() - opened = true - Log.i(TAG, "ffmpeg demuxer ready: ${trackOutputs.count { it != null }} tracks") - return true - } - FfmpegDemuxerJni.CODE_NEED_SEEK -> { - val target = FfmpegDemuxerJni.nativeConsumePendingSeek() - Log.i(TAG, "open: NEED_SEEK target=$target inputPos=${input.position}") - if (target == Long.MIN_VALUE || target < 0) { - // Sticky AVIO error replayed without a fresh deferral: re-sync at - // the loader's current position and continue (bounded below). - if (++reconciles > MAX_RECONCILES) { - throw malformed("demuxer produced an invalid seek target: $target") - } - if (FfmpegDemuxerJni.nativeResumeAfterSeek(input.position) != 0) { - throw malformed("demuxer resync failed at ${input.position}") - } - continue - } - if (++reconciles > MAX_RECONCILES) throw malformed("too many demuxer repositions") - if (!reconcile(input, seekPosition, target)) { - if (++openLoaderRoundTrips > MAX_OPEN_LOADER_ROUND_TRIPS) { - throw malformed("demuxer open did not converge after $openLoaderRoundTrips loader seeks") - } - return false - } - } - FfmpegDemuxerJni.ERR_JAVA -> - throw IOException(inputProxy.lastError ?: "demuxer input failed") - else -> throw malformed("ffmpeg demuxer open failed: $code") + private fun openStreams() { + when (val code = FfmpegDemuxerJni.nativeOpen(inputProxy)) { + 0 -> { + prepareTracks() + opened = true + Log.i(TAG, "ffmpeg demuxer ready: ${trackOutputs.count { it != null }} tracks") } + FfmpegDemuxerJni.ERR_JAVA -> throw IOException(inputProxy.lastError ?: "demuxer input failed") + else -> throw malformed("ffmpeg demuxer open failed: $code") } } @@ -435,12 +458,6 @@ internal class FfmpegExtractor private constructor( trackOutputs[index] = trackOutput } - primaryStreamIndex = if (primaryVideo >= 0) primaryVideo else primaryAudio - primaryStreamIsAudio = primaryVideo < 0 && primaryAudio >= 0 - synchronized(seekIndexLock) { - seekIndexTimes.clear() - seekIndexPositions.clear() - } output.endTracks() output.seekMap(SeekMapImpl()) } @@ -468,8 +485,7 @@ internal class FfmpegExtractor private constructor( } } - private fun readPacket(input: ExtractorInput, seekPosition: PositionHolder): Int { - var reconciles = 0 + private fun readPacket(): Int { while (true) { val code = FfmpegDemuxerJni.nativeReadPacket(packetBytes, packetOut) when (code) { @@ -521,24 +537,11 @@ internal class FfmpegExtractor private constructor( /* cryptoData= */ null ) - recordSeekPoint(streamIndex, ptsUs, packetOut[FfmpegDemuxerJni.OUT_POSITION], isKeyframe) + deliveredAnyPacket = true bitrateMeters.getOrNull(streamIndex)?.onPacket(ptsUs, size) return Extractor.RESULT_CONTINUE } FfmpegDemuxerJni.CODE_EOF -> return Extractor.RESULT_END_OF_INPUT - FfmpegDemuxerJni.CODE_NEED_SEEK -> { - val target = FfmpegDemuxerJni.nativeConsumePendingSeek() - if (target == Long.MIN_VALUE || target < 0) { - // Sticky AVIO error replayed without a fresh deferral: resync in - // place (bounded by reconciles) instead of killing the stream. - if (++reconciles > MAX_RECONCILES || !reSync(input)) { - throw malformed("demuxer produced an invalid seek target: $target") - } - continue - } - if (++reconciles > MAX_RECONCILES) throw malformed("too many demuxer repositions") - if (!reconcile(input, seekPosition, target)) return Extractor.RESULT_SEEK - } FfmpegDemuxerJni.CODE_GROW -> { val required = packetOut[FfmpegDemuxerJni.OUT_SIZE].toInt() packetBytes = ByteArray(required + 64 * 1024) @@ -550,81 +553,22 @@ internal class FfmpegExtractor private constructor( } } - /** Re-establishes the AVIO position at the loader's current input position. */ - private fun reSync(input: ExtractorInput): Boolean { - val position = input.position - return position >= 0 && FfmpegDemuxerJni.nativeResumeAfterSeek(position) == 0 - } - - /** Returns true when the input is already at [target] and native resumed. */ - private fun reconcile(input: ExtractorInput, seekPosition: PositionHolder, target: Long): Boolean { - if (target == input.position) { - val code = FfmpegDemuxerJni.nativeResumeAfterSeek(target) - if (code != 0) throw malformed("demuxer resume failed: $code") - return true - } - seekPosition.position = target - return false - } - - private fun recordSeekPoint(streamIndex: Int, timeUs: Long, position: Long, keyframe: Boolean) { - if (streamIndex != primaryStreamIndex || position <= 0) return - if (!primaryStreamIsAudio && !keyframe) return - synchronized(seekIndexLock) { - val lastIndex = seekIndexTimes.size - 1 - if (lastIndex >= 0) { - if (timeUs <= seekIndexTimes[lastIndex]) return - if (primaryStreamIsAudio && timeUs - seekIndexTimes[lastIndex] < AUDIO_INDEX_INTERVAL_US) { - return - } - } - seekIndexTimes.add(timeUs) - seekIndexPositions.add(position) - } - } - private fun malformed(message: String): ParserException = ParserException.createForMalformedContainer(message, null) private inner class SeekMapImpl : SeekMap { - // Seeks are executed by the demuxer (avformat_seek_file), so seekability - // does not depend on the sample-derived index below — that index only - // improves the loader's initial byte guess. The timeline is published - // from this value once, at prepare time, when the index is still empty. override fun isSeekable(): Boolean = this@FfmpegExtractor.durationUs != C.TIME_UNSET - override fun getSeekPoints(timeUs: Long): SeekMap.SeekPoints { - synchronized(seekIndexLock) { - if (seekIndexTimes.isEmpty()) return SeekMap.SeekPoints(SeekPoint(0, 0)) - val index = seekIndexTimes.binarySearchFloor(timeUs) - if (index < 0) return SeekMap.SeekPoints(SeekPoint(seekIndexTimes[0], seekIndexPositions[0])) - val preceding = SeekPoint(seekIndexTimes[index], seekIndexPositions[index]) - if (index + 1 >= seekIndexTimes.size || seekIndexTimes[index] == timeUs) { - return SeekMap.SeekPoints(preceding) - } - val following = SeekPoint(seekIndexTimes[index + 1], seekIndexPositions[index + 1]) - return SeekMap.SeekPoints(preceding, following) - } - } + /** + * The demuxer owns seek resolution: it reads its own index and lands on + * the right keyframe by itself, so there is no second index here to keep + * consistent. The byte position exists only to satisfy the interface — + * [alignLoader] moves the loader to wherever libavformat actually ended + * up, on the read that follows the seek. + */ + override fun getSeekPoints(timeUs: Long): SeekMap.SeekPoints = SeekMap.SeekPoints(SeekPoint(timeUs, 0)) // Qualified on purpose: bare `durationUs` binds to the inherited Java // getter as a synthetic property (this.getDurationUs()) and recurses. override fun getDurationUs(): Long = this@FfmpegExtractor.durationUs } } - -/** Floor binary search over an ascending list; returns the greatest index with value <= target. */ -private fun List.binarySearchFloor(value: Long): Int { - var low = 0 - var high = size - 1 - var result = -1 - while (low <= high) { - val mid = (low + high) ushr 1 - if (this[mid] <= value) { - result = mid - low = mid + 1 - } else { - high = mid - 1 - } - } - return result -} diff --git a/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/FfmpegRandomAccessSource.kt b/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/FfmpegRandomAccessSource.kt new file mode 100644 index 000000000..2a7227ef0 --- /dev/null +++ b/android/app/src/main/kotlin/com/edde746/plezy/exoplayer/FfmpegRandomAccessSource.kt @@ -0,0 +1,167 @@ +package com.edde746.plezy.exoplayer + +import android.net.Uri +import androidx.annotation.OptIn +import androidx.media3.common.C +import androidx.media3.common.util.UnstableApi +import androidx.media3.datasource.DataSource +import androidx.media3.datasource.DataSpec +import androidx.media3.datasource.TransferListener +import java.io.IOException + +/** + * Random-access byte source for the libavformat demuxer. + * + * libavformat demuxers own their own IO: they jump to the end of the file for an + * index (matroska Cues, MP4 moov, AVI idx1), back to the header, and binary + * search (MPEG-PS) whenever they please. media3's [ + * androidx.media3.extractor.ExtractorInput] is forward-only — an extractor may + * only ask the loader to re-open elsewhere and be invoked again — so serving + * libavformat from the loader alone means every backward jump has to unwind the + * in-flight libavformat call and replay it, which is what the demuxer shim used + * to do (and what made a matroska Cues read unrecoverable, #2096). + * + * This class removes the mismatch instead of working around it: reads outside + * the loader's window open a second [DataSource] from the same factory media3 + * uses, positioned exactly where libavformat asked. Cronet, download caches, + * SAF and the item's own request headers therefore still apply, because the + * [DataSpec] is cloned from the one media3 opened for this item. + * + * Sequential sample delivery deliberately does not come through here: it stays + * on the loader's connection so media3 keeps its byte accounting, back-pressure + * and load-error policy (see [FfmpegExtractor]). This source is opened for + * header/index bursts and closed again as soon as the loader is back in step. + */ +@OptIn(UnstableApi::class) +internal class FfmpegRandomAccessSource( + private val factory: DataSource.Factory, + private val specProvider: () -> DataSpec? +) { + private var source: DataSource? = null + + /** Absolute position the open handle returns next; [C.INDEX_UNSET] when closed. */ + private var handlePosition = C.INDEX_UNSET.toLong() + + private var resolvedLength = C.LENGTH_UNSET.toLong() + + private var opens = 0 + + /** Handles opened so far; the tests assert an index seek does not walk the file. */ + val openCount: Int + get() = synchronized(this) { opens } + + val isOpen: Boolean + get() = synchronized(this) { source != null } + + /** + * Reads up to [length] bytes at absolute [position]. Returns the number of + * bytes read, or 0 at end of input. Reopens the underlying source whenever + * the requested position is not where the current handle stands, which is the + * only way to move backwards on a byte stream. + */ + fun readAt(position: Long, buffer: ByteArray, length: Int): Int = synchronized(this) { + if (length <= 0) return 0 + val handle = handleAt(position) + val read = handle.read(buffer, 0, length) + if (read == C.RESULT_END_OF_INPUT) return 0 + handlePosition = position + read + read + } + + /** + * Total resource length once some open has resolved it, else + * [C.LENGTH_UNSET]. Never performs IO of its own: the extractor prefers the + * length media3 already resolved for the loader. + */ + fun length(): Long = synchronized(this) { resolvedLength } + + fun close() = synchronized(this) { + closeHandle() + handlePosition = C.INDEX_UNSET.toLong() + } + + private fun handleAt(position: Long): DataSource { + val existing = source + if (existing != null && handlePosition == position) return existing + closeHandle() + val spec = specProvider() ?: throw IOException("ffmpeg demuxer has no data spec for random access") + // Clone rather than subrange: the recorded spec carries the loader's own + // position, and subrange() is relative to it. + val positioned = spec.buildUpon() + .setPosition(position) + .setLength(C.LENGTH_UNSET.toLong()) + .build() + val created = factory.createDataSource() + val remaining = try { + created.open(positioned) + } catch (e: Throwable) { + closeQuietly(created) + throw e + } + source = created + handlePosition = position + opens++ + if (remaining != C.LENGTH_UNSET.toLong()) resolvedLength = position + remaining + return created + } + + private fun closeHandle() { + source?.let { closeQuietly(it) } + source = null + } + + private fun closeQuietly(dataSource: DataSource) { + try { + dataSource.close() + } catch (_: IOException) { + // Closing a handle we are abandoning cannot fail the read that replaces it. + } + } +} + +/** + * [DataSource.Factory] that remembers the [DataSpec] media3 opened for each + * item, so [FfmpegRandomAccessSource] can clone it — same URI, cache key, flags + * and request headers — instead of re-deriving one and silently losing, say, a + * download's cache key. + * + * Keyed by URI because ExoPlayer prepares the next item's period while the + * current one is still loading; the demuxer must never read the next item's + * bytes. + */ +@OptIn(UnstableApi::class) +internal class RecordingDataSourceFactory(private val delegate: DataSource.Factory) : DataSource.Factory { + + private val specs = object : LinkedHashMap(8, 0.75f, false) { + override fun removeEldestEntry(eldest: MutableMap.MutableEntry): Boolean = size > MAX_TRACKED_ITEMS + } + + fun specFor(uri: Uri): DataSpec? = synchronized(specs) { specs[uri] } + + override fun createDataSource(): DataSource = RecordingDataSource(delegate.createDataSource()) + + private fun record(dataSpec: DataSpec) { + synchronized(specs) { specs[dataSpec.uri] = dataSpec } + } + + private inner class RecordingDataSource(private val inner: DataSource) : DataSource { + override fun open(dataSpec: DataSpec): Long { + record(dataSpec) + return inner.open(dataSpec) + } + + override fun read(buffer: ByteArray, offset: Int, length: Int): Int = inner.read(buffer, offset, length) + + override fun addTransferListener(transferListener: TransferListener) = inner.addTransferListener(transferListener) + + override fun getUri(): Uri? = inner.uri + + override fun getResponseHeaders(): Map> = inner.responseHeaders + + override fun close() = inner.close() + } + + private companion object { + const val MAX_TRACKED_ITEMS = 4 + } +} diff --git a/android/app/src/test/kotlin/com/edde746/plezy/exoplayer/FfmpegRandomAccessSourceTest.kt b/android/app/src/test/kotlin/com/edde746/plezy/exoplayer/FfmpegRandomAccessSourceTest.kt new file mode 100644 index 000000000..7a7cd9e0e --- /dev/null +++ b/android/app/src/test/kotlin/com/edde746/plezy/exoplayer/FfmpegRandomAccessSourceTest.kt @@ -0,0 +1,222 @@ +package com.edde746.plezy.exoplayer + +import android.net.Uri +import androidx.media3.common.C +import androidx.media3.datasource.DataSource +import androidx.media3.datasource.DataSpec +import androidx.media3.datasource.TransferListener +import java.io.IOException +import org.junit.Assert.assertArrayEquals +import org.junit.Assert.assertEquals +import org.junit.Assert.assertFalse +import org.junit.Assert.assertNull +import org.junit.Assert.assertSame +import org.junit.Assert.assertThrows +import org.junit.Assert.assertTrue +import org.junit.Test +import org.junit.runner.RunWith +import org.robolectric.RobolectricTestRunner + +/** + * The demuxer's correctness now rests on this source answering any absolute + * position, so these pin the contract libavformat depends on: a read lands on + * the requested byte, a backward jump reopens rather than returning the wrong + * bytes, and sequential reads do not reopen (which would turn steady-state + * playback into one request per 64 KB). + */ +@RunWith(RobolectricTestRunner::class) +class FfmpegRandomAccessSourceTest { + + private val content = ByteArray(4096) { (it % 251).toByte() } + private val uri: Uri = Uri.parse("https://example.test/movie.mkv") + + private class FakeDataSource(private val content: ByteArray) : DataSource { + var openedAt: Long = -1 + var closed = false + private var position = 0 + + override fun open(dataSpec: DataSpec): Long { + openedAt = dataSpec.position + position = dataSpec.position.toInt() + return (content.size - position).toLong() + } + + override fun read(buffer: ByteArray, offset: Int, length: Int): Int { + if (position >= content.size) return C.RESULT_END_OF_INPUT + val count = minOf(length, content.size - position) + content.copyInto(buffer, offset, position, position + count) + position += count + return count + } + + override fun addTransferListener(transferListener: TransferListener) = Unit + + override fun getUri(): Uri? = null + + override fun getResponseHeaders(): Map> = emptyMap() + + override fun close() { + closed = true + } + } + + private inner class FakeFactory : DataSource.Factory { + val created = mutableListOf() + + override fun createDataSource(): DataSource = FakeDataSource(content).also { created.add(it) } + } + + private fun sourceFor(factory: DataSource.Factory, spec: DataSpec? = DataSpec(uri)) = FfmpegRandomAccessSource(factory) { spec } + + @Test + fun readsLandOnTheRequestedAbsolutePosition() { + val factory = FakeFactory() + val source = sourceFor(factory) + val buffer = ByteArray(16) + + assertEquals(16, source.readAt(1000, buffer, 16)) + + assertArrayEquals(content.copyOfRange(1000, 1016), buffer) + assertEquals(1000L, factory.created.single().openedAt) + } + + @Test + fun sequentialReadsReuseOneHandle() { + val factory = FakeFactory() + val source = sourceFor(factory) + val buffer = ByteArray(64) + + source.readAt(0, buffer, 64) + source.readAt(64, buffer, 64) + source.readAt(128, buffer, 64) + + assertEquals(1, source.openCount) + assertArrayEquals(content.copyOfRange(128, 192), buffer) + } + + @Test + fun backwardJumpReopensAndReturnsTheRequestedBytes() { + val factory = FakeFactory() + val source = sourceFor(factory) + val buffer = ByteArray(32) + + source.readAt(2048, buffer, 32) + // The jump libavformat makes for an index at the end of the file, then back. + assertEquals(32, source.readAt(16, buffer, 32)) + + assertArrayEquals(content.copyOfRange(16, 48), buffer) + assertEquals(2, source.openCount) + assertTrue("abandoned handle must be closed", factory.created.first().closed) + } + + @Test + fun endOfInputReportsZeroWithoutAdvancing() { + val factory = FakeFactory() + val source = sourceFor(factory) + val buffer = ByteArray(8) + + assertEquals(0, source.readAt(content.size.toLong(), buffer, 8)) + } + + @Test + fun lengthResolvesFromAnOpenAndStaysKnown() { + val factory = FakeFactory() + val source = sourceFor(factory) + + assertEquals(C.LENGTH_UNSET.toLong(), source.length()) + source.readAt(512, ByteArray(8), 8) + + assertEquals(content.size.toLong(), source.length()) + } + + @Test + fun clonedSpecKeepsCacheKeyAndFlagsButTakesOurPosition() { + val recorded = DataSpec.Builder() + .setUri(uri) + .setPosition(9999) + .setLength(1234) + .setKey("download-cache-key") + .setFlags(DataSpec.FLAG_ALLOW_CACHE_FRAGMENTATION) + .setHttpRequestHeaders(mapOf("X-Test" to "1")) + .build() + var openedSpec: DataSpec? = null + val factory = DataSource.Factory { + object : DataSource { + override fun open(dataSpec: DataSpec): Long { + openedSpec = dataSpec + return C.LENGTH_UNSET.toLong() + } + + override fun read(buffer: ByteArray, offset: Int, length: Int): Int = C.RESULT_END_OF_INPUT + + override fun addTransferListener(transferListener: TransferListener) = Unit + + override fun getUri(): Uri? = null + + override fun getResponseHeaders(): Map> = emptyMap() + + override fun close() = Unit + } + } + + sourceFor(factory, recorded).readAt(4242, ByteArray(4), 4) + + val spec = requireNotNull(openedSpec) + assertEquals(4242L, spec.position) + assertEquals(C.LENGTH_UNSET.toLong(), spec.length) + assertEquals("download-cache-key", spec.key) + assertEquals(DataSpec.FLAG_ALLOW_CACHE_FRAGMENTATION, spec.flags) + assertEquals("1", spec.httpRequestHeaders["X-Test"]) + assertEquals(uri, spec.uri) + } + + @Test + fun readWithoutARecordedSpecFails() { + val source = sourceFor(FakeFactory(), spec = null) + + assertThrows(IOException::class.java) { source.readAt(0, ByteArray(4), 4) } + } + + @Test + fun closeReleasesTheHandle() { + val factory = FakeFactory() + val source = sourceFor(factory) + source.readAt(0, ByteArray(4), 4) + + assertTrue(source.isOpen) + source.close() + + assertFalse(source.isOpen) + assertTrue(factory.created.single().closed) + } + + @Test + fun recordingFactoryReportsTheSpecMedia3OpenedPerUri() { + val delegate = FakeFactory() + val recorder = RecordingDataSourceFactory(delegate) + val other = Uri.parse("https://example.test/next-episode.mkv") + + assertNull(recorder.specFor(uri)) + recorder.createDataSource().open(DataSpec.Builder().setUri(uri).setPosition(64).setKey("a").build()) + recorder.createDataSource().open(DataSpec.Builder().setUri(other).setPosition(128).build()) + + // Preloading the next item must not redirect the current demuxer's reads. + assertEquals(64L, recorder.specFor(uri)?.position) + assertEquals("a", recorder.specFor(uri)?.key) + assertEquals(128L, recorder.specFor(other)?.position) + } + + @Test + fun recordingFactoryDelegatesReads() { + val delegate = FakeFactory() + val recorder = RecordingDataSourceFactory(delegate) + val buffer = ByteArray(16) + + val dataSource = recorder.createDataSource() + dataSource.open(DataSpec.Builder().setUri(uri).setPosition(32).build()) + assertEquals(16, dataSource.read(buffer, 0, 16)) + + assertArrayEquals(content.copyOfRange(32, 48), buffer) + assertSame(delegate.created.single(), delegate.created.first()) + } +}