From c88778002202870d499ae2fa6a54dbf9abd36713 Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 21 Oct 2026 00:13:01 +0200 Subject: [PATCH] Consume Android batches with independent track completion --- .../com/utun/chat/data/StreamPlayerEngine.kt | 47 ++++++++++++---- .../com/utun/chat/data/StreamReadQueue.kt | 47 ++++++++++++++++ .../com/utun/chat/data/StreamReadQueueTest.kt | 55 +++++++++++++++++++ 3 files changed, 139 insertions(+), 10 deletions(-) create mode 100644 tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamReadQueue.kt create mode 100644 tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamReadQueueTest.kt diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt index 0b535bea..27bbd487 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamPlayerEngine.kt @@ -101,6 +101,12 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv var audioEnabled = false var opened = false var ended = false + var videoInputEnded = false + var audioInputEnded = false + val records = StreamReadQueue() + var signal = 0L + var batches = 0L + var batchMax = 0 var videoRevision = 0L var position = -1L var ready = -1L @@ -235,7 +241,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv ready = meta[0] if (playback != null) return audioEnabled = meta[5] and 2 != 0L - playback = StreamMixedPlayback(id, meta[3], audioEnabled) + playback = StreamMixedPlayback(id, meta[3], audioEnabled) { NativeLib.streamSignalNotify(signal) } position = meta[3]; generation = meta[4] LogManager.addLog("INFO", "stream", "playback starts id=$id PTS=$position generation=$generation fixedPrefill=0 " + "elapsedUs=${System.nanoTime() / 1000 - requestedAtUs} readyUs=$ready videoUs=${meta[1]} audioUs=${meta[2]} " + @@ -253,22 +259,29 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv try { playback?.close() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio reset id=$id: ${e.message}") } playback = null; position = -1; generation = 0; needsAudio = true; clockState = null; playbackStarted = false + records.clear(); ended = false; videoInputEnded = false; audioInputEnded = false resetVideo(); deadline = System.nanoTime() + 10_000_000_000 if (request) check(NativeLib.streamResync(id)) { "Media resync refused" } } video = createVideoPump() try { check(NativeLib.streamOpen(id, group, source, audio = true, epoch = epoch)) { "Viewer open refused" } + signal = NativeLib.streamSignalCreate(id).also { check(it != 0L) { "Viewer notification initialization failed" } } while (running && surface.isValid) { check(NativeLib.isReady()) { "Core stopped" } observeStatus(NativeLib.streamStatus(id)) drainTimeoutMs = ((ready - position) / 1000 + 2000).coerceIn(2000, 30000) val deviceLead = playback?.leadUs ?: 0 - for (work in 0 until 4) { + var consumed = 0 + for (work in 0 until 32) { video.step(position, System.nanoTime() / 1000) - val audioAhead = if (audioEnabled && needsAudio) deviceLead + 80000 else -1L - val raw = if (ended) null else NativeLib.streamPull(id, position, video.wantsInput, audioAhead) - if (raw == null) break + needsAudio = !audioInputEnded && (playback?.needsInput ?: true) + val audioAhead = if (!audioInputEnded && (!audioEnabled || needsAudio) && records.audios == 0) deviceLead + 80000 else -1L + val videoBudget = if (!videoInputEnded && video.wantsInput && records.videos == 0) 1 else 0 + val batch = if (ended) null else NativeLib.streamPull(id, position, videoBudget, audioAhead, 16) + if (batch != null) { batches++; batchMax = maxOf(batchMax, records.add(batch)) } + val raw = records.take(!videoInputEnded && video.wantsInput, !audioEnabled || needsAudio) ?: break + consumed++ check(raw.size >= 28) val b = ByteBuffer.wrap(raw).order(ByteOrder.BIG_ENDIAN) val kind = b.get().toInt(); val flags = b.get().toInt() and 255 @@ -308,7 +321,13 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv } 3 -> { LogManager.addLog("INFO", "stream", "playback GAP id=$id"); resetTimeline(false) } 4 -> opened = true - 5 -> { ended = true; playback?.finish() } + 5, 7, 8 -> { + if (kind != 8) videoInputEnded = true + if (kind != 7 && !audioInputEnded) { audioInputEnded = true; playback?.finish() } + if (kind == 5) ended = true + LogManager.addLog("INFO", "stream", "viewer track end id=$id kind=$kind videoEnd=$videoInputEnded " + + "audioEnd=$audioInputEnded pending=${records.size}") + } 6 -> { mutableStatus.value = when (flags) { 1 -> R.string.stream_group_closed; 2 -> R.string.chat_service_stopped @@ -316,6 +335,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv else -> R.string.stream_failed } LogManager.addLog("WARN", "stream", "native viewer closed id=$id reason=$flags"); running = false + break } else -> error("Unknown media event=$kind") } @@ -333,12 +353,12 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv } val codec = decoder if (codec != null) { - if (ended && video.input == null && video.output == null && !eosQueued) { + if (videoInputEnded && video.input == null && video.output == null && !eosQueued) { val index = codec.dequeueInputBuffer(0) if (index >= 0) { codec.queueInputBuffer(index, 0, 0, video.lastInputPts.coerceAtLeast(0), MediaCodec.BUFFER_FLAG_END_OF_STREAM); eosQueued = true } } - video.step(position, System.nanoTime() / 1000); video.checkProgress(System.nanoTime() / 1000, ended) + video.step(position, System.nanoTime() / 1000); video.checkProgress(System.nanoTime() / 1000, videoInputEnded) check(!video.ended || video.firstOutputPts != null || video.inputCount == 0L) { "Decoder ended without producing an image" } } if (ended && video.input == null && (decoder == null || video.ended) && playback?.empty != false @@ -354,9 +374,15 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv "clock=$position ready=$ready audio=$audioEnabled audioInputs=$audioInputs audioPTS=$lastAudioPts needsAudio=$needsAudio " + "reserve=${clockState?.get(1)} minimum=${clockState?.get(2)} " + "tempo=${clockState?.get(3)} ${playback?.description} waitingKey=$waitingKey " + - "shownPTS=$lastShownPts displayLagUs=${if (lastShownPts >= 0) position - lastShownPts else -1}") + "shownPTS=$lastShownPts displayLagUs=${if (lastShownPts >= 0) position - lastShownPts else -1} " + + "batches=$batches batchMax=$batchMax pending=${records.size} videoEnd=$videoInputEnded audioEnd=$audioInputEnded") + batches = 0; batchMax = 0 + } + // Синхронный MediaCodec требует проверки готовности; media/PCM/stop будят без ожидания deadline. + if (running && consumed < 32) { + val codecPending = video.input != null || video.output != null || video.inflight > 0 || (videoInputEnded && !video.ended) + check(NativeLib.streamSignalWait(signal, if (codecPending) 5 else 50) >= 0) { "Viewer notification wait failed" } } - Thread.sleep(5) } } catch (e: Exception) { if (running) { @@ -369,6 +395,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder release id=$id: ${e.message}") } try { playback?.close() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio release id=$id: ${e.message}") } + NativeLib.streamSignalClose(signal) running = false LogManager.addLog("INFO", "stream", "viewer released id=$id inputs=${video.inputCount} " + "decoded=${video.outputCount} shown=${video.shown} discarded=${video.discarded}") diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamReadQueue.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamReadQueue.kt new file mode 100644 index 00000000..51419205 --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamReadQueue.kt @@ -0,0 +1,47 @@ +package com.utun.chat.data + +import java.nio.ByteBuffer +import java.nio.ByteOrder + +/** Ближайшие записи viewer: порядок каждого трека независим, EOF не ждёт чужой трек. + * GAP/CLOSED отменяют прежние данные. Только player worker; ошибки логирует его владелец. */ +internal class StreamReadQueue { + private val records = mutableListOf() + val size: Int get() = records.size + val videos: Int get() = records.count { it[0].toInt() == 1 } + val audios: Int get() = records.count { it[0].toInt() == 2 } + fun clear() = records.clear() + + fun add(batch: ByteArray): Int { + val input = ByteBuffer.wrap(batch).order(ByteOrder.BIG_ENDIAN) + var count = 0 + while (input.hasRemaining()) { + check(input.remaining() >= 28) { "Truncated stream record header" } + val start = input.position() + val kind = batch[start].toInt() + val length = input.getInt(start + 16) + check(kind in 1..8 && length >= 0 && length <= input.remaining() - 28) { "Invalid stream record kind=$kind length=$length" } + if (kind == 3 || kind == 6) records.clear() + records.add(batch.copyOfRange(start, start + 28 + length)) + input.position(start + 28 + length) + check(++count <= 16) { "Stream batch exceeds record budget" } + } + return count + } + + fun take(video: Boolean, audio: Boolean): ByteArray? { + var videoBefore = false + var audioBefore = false + for (i in records.indices) { + val eligible = when (records[i][0].toInt()) { + 1 -> video.also { videoBefore = true } + 2 -> audio.also { audioBefore = true } + 7 -> !videoBefore + 8 -> !audioBefore + else -> i == 0 + } + if (eligible) return records.removeAt(i) + } + return null + } +} diff --git a/tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamReadQueueTest.kt b/tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamReadQueueTest.kt new file mode 100644 index 00000000..33997a5f --- /dev/null +++ b/tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamReadQueueTest.kt @@ -0,0 +1,55 @@ +package com.utun.chat.data + +import org.junit.Assert.* +import org.junit.Test +import java.nio.ByteBuffer + +class StreamReadQueueTest { + private fun record(kind: Int, pts: Long = 0, bytes: ByteArray = byteArrayOf()): ByteArray = + ByteBuffer.allocate(28 + bytes.size).apply { + put(0, kind.toByte()); putLong(8, pts); putInt(16, bytes.size); position(28); put(bytes) + }.array() + + @Test fun videoEndPassesBlockedAudioButPreservesVideoOrder() { + val queue = StreamReadQueue() + queue.add(record(2, 20) + record(1, 40) + record(7) + record(8) + record(5)) + assertNull(queue.take(video = false, audio = false)) + assertEquals(1, queue.take(video = true, audio = false)!![0].toInt()) + assertEquals(7, queue.take(video = false, audio = false)!![0].toInt()) + assertNull(queue.take(video = false, audio = false)) + assertEquals(2, queue.take(video = false, audio = true)!![0].toInt()) + assertEquals(8, queue.take(video = false, audio = false)!![0].toInt()) + assertEquals(5, queue.take(video = false, audio = false)!![0].toInt()) + assertEquals(0, queue.size) + } + + @Test fun audioEndPassesBlockedVideo() { + val queue = StreamReadQueue() + queue.add(record(1) + record(2) + record(8) + record(7) + record(5)) + assertEquals(2, queue.take(video = false, audio = true)!![0].toInt()) + assertEquals(8, queue.take(video = false, audio = false)!![0].toInt()) + assertNull(queue.take(video = false, audio = false)) + assertEquals(1, queue.take(video = true, audio = false)!![0].toInt()) + assertEquals(7, queue.take(video = false, audio = false)!![0].toInt()) + assertEquals(5, queue.take(video = false, audio = false)!![0].toInt()) + } + + @Test fun gapAndCloseImmediatelyCancelBlockedTracks() { + for (kind in listOf(3, 6)) { + val queue = StreamReadQueue() + queue.add(record(1) + record(2)) + queue.add(record(kind)) + assertEquals(kind, queue.take(video = false, audio = false)!![0].toInt()) + assertEquals(0, queue.size) + } + } + + @Test fun variablePayloadsKeepRecordBoundaries() { + val queue = StreamReadQueue() + val video = record(1, 42, byteArrayOf(1, 2, 3)) + val audio = record(2, 43, byteArrayOf(4)) + assertEquals(2, queue.add(video + audio)) + assertArrayEquals(audio, queue.take(video = false, audio = true)) + assertArrayEquals(video, queue.take(video = true, audio = false)) + } +}