From 0fdc6e435b3a06a257fc738b2ddbc0f172d7f56e Mon Sep 17 00:00:00 2001 From: evgeny Date: Wed, 14 Oct 2026 17:16:45 +0200 Subject: [PATCH] Fix Android stream audio backpressure without blocking video --- .../com/utun/chat/data/StreamAudioOutput.kt | 143 ++++++++++++++++++ .../com/utun/chat/data/StreamPlayerEngine.kt | 79 ++-------- .../utun/chat/data/StreamAudioOutputTest.kt | 133 ++++++++++++++++ 3 files changed, 290 insertions(+), 65 deletions(-) create mode 100644 tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioOutput.kt create mode 100644 tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamAudioOutputTest.kt diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioOutput.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioOutput.kt new file mode 100644 index 00000000..fc8f50e0 --- /dev/null +++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/StreamAudioOutput.kt @@ -0,0 +1,143 @@ +package com.utun.chat.data + +import android.media.AudioAttributes +import android.media.AudioFormat +import android.media.AudioTrack +import android.os.Build + +/** Один player worker владеет устройством и единственным недописанным PCM. + * Native уже учёл выданные кадры: остаток нельзя терять или переставлять до сброса всей шкалы. */ +internal class StreamAudioOutput( + private val open: () -> Sink = { AndroidSink() }, + private val log: (String) -> Unit = {} +) { + internal interface Sink { + val threshold: Int + val head: Long + val description: String + fun write(pcm: ShortArray, offset: Int, count: Int): Int + fun pause() + fun play() + fun close() + } + + private class AndroidSink : Sink { + private val track = AudioTrack.Builder() + .setAudioAttributes(AudioAttributes.Builder().setUsage(AudioAttributes.USAGE_MEDIA) + .setContentType(AudioAttributes.CONTENT_TYPE_MOVIE).build()) + .setAudioFormat(AudioFormat.Builder().setSampleRate(48000).setChannelMask(AudioFormat.CHANNEL_OUT_STEREO) + .setEncoding(AudioFormat.ENCODING_PCM_16BIT).build()) + .setBufferSizeInBytes(AudioTrack.getMinBufferSize(48000, AudioFormat.CHANNEL_OUT_STEREO, + AudioFormat.ENCODING_PCM_16BIT).coerceAtLeast(3840)).setTransferMode(AudioTrack.MODE_STREAM).build() + init { + try { + check(track.state == AudioTrack.STATE_INITIALIZED) { "AudioTrack initialization failed" } + // API 26..30 не умеет уменьшать порог отдельно от аппаратного буфера. + if (Build.VERSION.SDK_INT >= 31) track.setStartThresholdInFrames(1) + } catch (e: Exception) { track.release(); throw e } + } + override val threshold: Int get() = if (Build.VERSION.SDK_INT >= 31) + minOf(track.startThresholdInFrames, track.bufferSizeInFrames) else track.bufferSizeInFrames + override val head: Long get() = track.playbackHeadPosition.toLong() and 0xffffffffL + override val description: String get() = "state=${track.state} playState=${track.playState} " + + "buffer=${track.bufferSizeInFrames} threshold=$threshold underruns=${track.underrunCount}" + override fun write(pcm: ShortArray, offset: Int, count: Int) = track.write(pcm, offset, count, AudioTrack.WRITE_NON_BLOCKING) + override fun pause() = track.pause() + override fun play() = track.play() + override fun close() = track.release() + } + + private var sink: Sink? = null + private var pending: ShortArray? = null + private var offset = 0 + private var padding = false + private var padded = 0L + private var previousHead = 0L + private var headEpoch = 0L + private var deviceHead = 0L + private var stalledSince = -1L + private var openedAt = 0L + private var paused = false + private val silence = ShortArray(960) + var written = 0L + private set + var played = 0L + private set + val pendingFrames: Int get() = if (padding) 0 else ((pending?.size ?: 0) - offset) / 2 + val leadUs: Long get() = (written - played + pendingFrames) * 1_000_000 / 48000 + val drained: Boolean get() = pendingFrames == 0 && played == written + val description: String get() = "written=$written played=$played pending=$pendingFrames padded=$padded " + + "head=$deviceHead paused=$paused ${sink?.description ?: "device=closed"}" + + /** Не экстраполируем старый timestamp: при остановке устройства media clock также стоит. */ + private fun poll(now: Long) { + val s = sink ?: return + val raw = s.head + if (raw < previousHead) { + check(previousHead - raw > 0x80000000L) { "AudioTrack head regressed $previousHead->$raw; $description" } + headEpoch += 0x100000000L + } + previousHead = raw + val head = headEpoch + raw + if (head > deviceHead) { + if (deviceHead == 0L) log("audio started delayMs=${(now - openedAt) / 1_000_000} $description") + stalledSince = -1 + } + deviceHead = head + played = head.coerceAtMost(written) + } + + /** Ровно одна неблокирующая запись за итерацию; ожидание устройства не задерживает video pump. */ + fun step(now: Long, hold: Boolean, pull: () -> ShortArray?, finished: () -> Boolean) { + poll(now) + if (paused != hold) { + if (hold) sink?.pause() else sink?.play() + paused = hold; stalledSince = -1 + log("audio ${if (hold) "paused" else "resumed"} $description") + } + if (hold) return + if (pending == null && written - played < maxOf(960, sink?.threshold ?: 1)) { + val pcm = pull() + if (pcm != null) { + check(pcm.isNotEmpty() && pcm.size <= 960 && pcm.size % 2 == 0) { "Invalid stream PCM size=${pcm.size}" } + if (sink == null) { + sink = open(); openedAt = now; sink!!.play() + log("audio opened ${sink!!.description}") + } + pending = pcm; offset = 0; padding = false + } + } + val s = sink ?: return + // Короткий EOF должен пересечь порог старых AudioTrack даже после underrun. + // Тишина идёт только после всего PCM; played ограничен written и не сдвигает media PTS. + if (pending == null && finished() && played < written && padded < s.threshold) { + val frames = minOf(480L, s.threshold - padded).toInt() + pending = if (frames == 480) silence else ShortArray(frames * 2) + offset = 0; padding = true + if (padded == 0L) log("audio draining threshold=${s.threshold} $description") + } + var blocked = false + pending?.let { pcm -> + val count = s.write(pcm, offset, pcm.size - offset) + check(count >= 0 && count <= pcm.size - offset && count % 2 == 0) { "AudioTrack write=$count; $description" } + blocked = count == 0 + offset += count + if (padding) padded += count / 2 else written += count / 2 + if (offset == pcm.size) { pending = null; offset = 0; padding = false } + } + poll(now) + // До достаточного заполнения ожидание сети не является зависанием устройства. + val expectedProgress = written + padded - deviceHead >= s.threshold || blocked + if (!expectedProgress) stalledSince = -1 + else if (stalledSince < 0) stalledSince = now + else check(now - stalledSince < 2_000_000_000) { + "AudioTrack made no playback progress for ${(now - stalledSince) / 1_000_000}ms; $description" + } + } + + fun close() { + val s = sink + sink = null; pending = null; offset = 0; padding = false + s?.close() + } +} 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 b4ebb537..713d208e 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 @@ -1,9 +1,5 @@ package com.utun.chat.data -import android.media.AudioAttributes -import android.media.AudioFormat -import android.media.AudioTimestamp -import android.media.AudioTrack import android.media.MediaCodec import android.media.MediaCodecInfo import android.media.MediaCodecList @@ -44,28 +40,24 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv private fun play(surface: Surface) { var decoder: MediaCodec? = null - var track: AudioTrack? = null + fun audioOutput() = StreamAudioOutput(log = { LogManager.addLog("INFO", "stream", "viewer id=$id $it") }) + var audio = audioOutput() var playback = 0L lateinit var video: StreamDecodePump var waitingKey = true var audioEnabled = false var opened = false var ended = false - var pausedAudio = false var videoRevision = 0L var position = -1L var ready = -1L var needsAudio = true - var written = 0L - var previousHead = 0L - var headEpoch = 0L var generation = 0L var playbackStarted = false var eosQueued = false var endUntil = 0L var reported = 0L var deadline = System.nanoTime() + 10_000_000_000 - val timestamp = AudioTimestamp() val info = MediaCodec.BufferInfo() var clockState: LongArray? = null @@ -134,57 +126,21 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv mutableStatus.value = R.string.stream_waiting_key } fun resetTimeline(request: Boolean) { - try { track?.pause(); track?.flush(); track?.release() } + try { audio.close() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio reset id=$id: ${e.message}") } - track = null; pausedAudio = false; written = 0; previousHead = 0; headEpoch = 0 + audio = audioOutput() if (playback != 0L) NativeLib.streamPlaybackClose(playback) playback = 0; position = -1; generation = 0; needsAudio = true; clockState = null; playbackStarted = false resetVideo(); deadline = System.nanoTime() + 10_000_000_000 if (request) check(NativeLib.streamResync(id)) { "Media resync refused" } } - fun playedFrames(now: Long): Long { - val t = track ?: return 0 - val raw = t.playbackHeadPosition.toLong() and 0xffffffffL - if (raw < previousHead) headEpoch += 0x100000000L - previousHead = raw - val head = headEpoch + raw - if (pausedAudio || !t.getTimestamp(timestamp)) return head.coerceAtMost(written) - var frames = timestamp.framePosition - if (frames <= 0xffffffffL) { - frames += headEpoch - if (frames - head > 0x80000000L) frames -= 0x100000000L - if (head - frames > 0x80000000L) frames += 0x100000000L - } - return (frames + (now - timestamp.nanoTime).coerceAtLeast(0) * 48000 / 1_000_000_000).coerceIn(0, written) - } - fun writePcm(pcm: ShortArray) { - if (track == null) { - track = AudioTrack.Builder().setAudioAttributes(AudioAttributes.Builder().setUsage(AudioAttributes.USAGE_MEDIA) - .setContentType(AudioAttributes.CONTENT_TYPE_MOVIE).build()).setAudioFormat(AudioFormat.Builder() - .setSampleRate(48000).setChannelMask(AudioFormat.CHANNEL_OUT_STEREO).setEncoding(AudioFormat.ENCODING_PCM_16BIT).build()) - .setBufferSizeInBytes(AudioTrack.getMinBufferSize(48000, AudioFormat.CHANNEL_OUT_STEREO, - AudioFormat.ENCODING_PCM_16BIT).coerceAtLeast(3840)).setTransferMode(AudioTrack.MODE_STREAM).build() - check(track!!.state == AudioTrack.STATE_INITIALIZED); track!!.play() - LogManager.addLog("INFO", "stream", "audio output id=$id bufferFrames=${track!!.bufferSizeInFrames} stereo=1") - } - val until = System.nanoTime() + 100_000_000 - var offset = 0 - while (running && offset < pcm.size) { - val count = track!!.write(pcm, offset, pcm.size - offset, AudioTrack.WRITE_NON_BLOCKING) - check(count >= 0 && count % 2 == 0) { "AudioTrack write=$count" } - if (count > 0) { offset += count; written += count / 2 } - else { check(System.nanoTime() < until) { "AudioTrack stalled" }; Thread.sleep(2) } - } - } video = createVideoPump() try { check(NativeLib.streamOpen(id, group, source, audio = true, epoch = epoch)) { "Viewer open refused" } while (running && surface.isValid) { check(NativeLib.isReady()) { "Core stopped" } observeStatus(NativeLib.streamStatus(id)) - val now = System.nanoTime() - val played = playedFrames(now) - val deviceLead = (written - played).coerceAtLeast(0) * 1_000_000 / 48000 + val deviceLead = audio.leadUs for (work in 0 until 4) { video.step(position, System.nanoTime() / 1000) val audioAhead = if (audioEnabled && needsAudio) deviceLead + 80000 else -1L @@ -245,24 +201,17 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv LogManager.addLog("INFO", "video", "viewer first decoded frame id=$id revision=$videoRevision " + "PTS=${video.firstOutputPts} inputs=${video.inputCount} decoderLeadUs=${video.decoderLeadUs}") } - if (playbackStarted && playback != 0L && audioEnabled && !pausedAudio) { - val head = playedFrames(System.nanoTime()) - val t = track - val prime = t == null || (head == 0L && written < t.bufferSizeInFrames) - if (prime || written - head < 960) { - NativeLib.streamPlaybackPull(playback, 480, if (ended) Long.MAX_VALUE else ready)?.let { writePcm(it) } - } - } if (playbackStarted && playback != 0L && ready >= 0) { val lateVideo = !video.ended && (video.firstOutputPts == null || (video.oldestSubmittedPts >= 0 && video.oldestSubmittedPts < position - 10000)) - if (lateVideo && track != null && !pausedAudio) { track!!.pause(); pausedAudio = true } - if (pausedAudio && !lateVideo) { track!!.play(); pausedAudio = false } - val head = playedFrames(System.nanoTime()) - val audioEmpty = audioEnabled && written > 0 && head >= written && !ended + if (audioEnabled) audio.step(System.nanoTime(), lateVideo, + pull = { NativeLib.streamPlaybackPull(playback, 480, if (ended) Long.MAX_VALUE else ready) }, + finished = { ended && NativeLib.streamPlaybackEmpty(playback) }) + val head = audio.played + val audioEmpty = audioEnabled && head >= audio.written && !ended clockState = NativeLib.streamPlaybackUpdate(playback, System.nanoTime() / 1000, head, if (ended) ready.coerceAtLeast(position) else ready, - maxOf(deviceLead + 10000, video.decoderLeadUs + video.framePeriodUs), lateVideo || audioEmpty) + maxOf(audio.leadUs + 10000, video.decoderLeadUs + video.framePeriodUs), lateVideo || audioEmpty) check(clockState != null) { "Playback timeline failed" } position = clockState!![0]; needsAudio = clockState!![5] != 0L } @@ -277,7 +226,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv check(!video.ended || video.firstOutputPts != null || video.inputCount == 0L) { "Decoder ended without producing an image" } } if (ended && video.input == null && (decoder == null || video.ended) && NativeLib.streamPlaybackEmpty(playback) - && playedFrames(System.nanoTime()) >= written) { + && audio.drained) { if (endUntil == 0L) endUntil = System.nanoTime() + 100_000_000 if (System.nanoTime() >= endUntil) { mutableStatus.value = R.string.stream_ended; break } } @@ -287,7 +236,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv "decoded=${video.outputCount} shown=${video.shown} discarded=${video.discarded} input=${video.input != null} " + "output=${video.output?.pts} inflight=${video.inflight} decoderLeadUs=${video.decoderLeadUs} " + "clock=$position reserve=${clockState?.get(1)} minimum=${clockState?.get(2)} " + - "tempo=${clockState?.get(3)} written=$written waitingKey=$waitingKey") + "tempo=${clockState?.get(3)} ${audio.description} waitingKey=$waitingKey") } Thread.sleep(5) } @@ -300,7 +249,7 @@ class StreamPlayerEngine(private val group: Long, private val source: Long, priv NativeLib.streamStop(id) try { decoder?.stop(); decoder?.release() } catch (e: Exception) { LogManager.addLog("WARN", "video", "decoder release id=$id: ${e.message}") } - try { track?.stop(); track?.release() } + try { audio.close() } catch (e: Exception) { LogManager.addLog("WARN", "stream", "audio release id=$id: ${e.message}") } if (playback != 0L) NativeLib.streamPlaybackClose(playback) val until = System.nanoTime() + 500_000_000 diff --git a/tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamAudioOutputTest.kt b/tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamAudioOutputTest.kt new file mode 100644 index 00000000..b027bb39 --- /dev/null +++ b/tools/chatgui-android/app/src/test/java/com/utun/chat/data/StreamAudioOutputTest.kt @@ -0,0 +1,133 @@ +package com.utun.chat.data + +import org.junit.Assert.* +import org.junit.Test +import java.util.ArrayDeque + +class StreamAudioOutputTest { + private class Device(override val threshold: Int = 1) : StreamAudioOutput.Sink { + override var head = 0L + override val description: String get() = "fake threshold=$threshold" + val samples = ArrayList() + var limit = 960 + var calls = 0 + var paused = false + var closed = false + override fun write(pcm: ShortArray, offset: Int, count: Int): Int { + check(!paused && !closed) + calls++ + if (limit < 0) return limit + val accepted = minOf(limit, count) + for (i in offset until offset + accepted) samples.add(pcm[i]) + return accepted + } + override fun pause() { paused = true } + override fun play() { paused = false } + override fun close() { closed = true } + } + private class Player(val device: Device = Device()) { + val input = ArrayDeque() + val logs = ArrayList() + val output = StreamAudioOutput({ device }, { logs.add(it) }) + var finished = false + var pulls = 0 + fun step(ms: Long, hold: Boolean = false) = output.step(ms * 1_000_000, hold, + { pulls++; input.pollFirst() }, { finished && input.isEmpty() }) + } + + @Test fun slowStartupDoesNotBlockOtherPlayerWork() { + val p = Player(); p.input.add(ShortArray(960) { it.toShort() }); p.device.limit = 0 + var videoSteps = 0 + for (ms in 0L..300L step 5) { p.step(ms); videoSteps++ } + assertEquals(61, videoSteps); assertEquals(61, p.device.calls); assertEquals(1, p.pulls) + assertEquals(480, p.output.pendingFrames); assertEquals(0L, p.output.played) + p.device.limit = 960; p.step(305); p.device.head = 480; p.step(310) + assertTrue(p.output.drained); assertTrue(p.logs.any { it.contains("audio started") }) + } + + @Test fun partialWritesPreserveEverySampleAndOnlyPullOnce() { + val p = Player(); val pcm = ShortArray(960) { it.toShort() }; p.input.add(pcm) + p.device.limit = 120 + repeat(8) { p.step(it * 5L) } + assertEquals(pcm.toList(), p.device.samples); assertEquals(1, p.pulls) + assertEquals(480L, p.output.written); assertEquals(0L, p.output.played) + assertEquals(0, p.output.pendingFrames); assertFalse(p.output.drained) + } + + @Test fun waitingForInitialNetworkDataIsNotDeviceStall() { + val p = Player(Device(3848)); p.input.add(ShortArray(200)); p.step(0); p.step(10000) + assertEquals(100L, p.output.written); assertEquals(0L, p.output.played) + } + + @Test fun clockDoesNotAdvanceWithWallTimeAndStalledDeviceFails() { + val p = Player(); p.input.add(ShortArray(960)); p.step(0); p.step(1999) + assertEquals(0L, p.output.played) + val error = assertThrows(IllegalStateException::class.java) { p.step(2000) } + assertTrue(error.message!!.contains("no playback progress")) + assertTrue(error.message!!.contains("written=480 played=0")) + } + + @Test fun actualHeadProgressRestartsWatchdog() { + val p = Player(); p.input.add(ShortArray(960)); p.step(0) + p.device.head = 100; p.step(1900); p.step(3800) + assertEquals(100L, p.output.played) + p.device.head = 480; p.step(3900); p.step(10000) + assertTrue(p.output.drained) + } + + @Test fun intentionalPausePreservesPartialPcmAndSuspendsWatchdog() { + val p = Player(); val pcm = ShortArray(960) { it.toShort() }; p.input.add(pcm); p.device.limit = 120 + p.step(0); p.step(5, true); p.step(10000, true) + assertTrue(p.device.paused); assertEquals(1, p.device.calls); assertEquals(420, p.output.pendingFrames) + p.device.limit = 960; p.step(10005) + assertFalse(p.device.paused); assertEquals(pcm.toList(), p.device.samples) + } + + @Test fun eofWaitsForPendingPcmAndPhysicalPlayback() { + val p = Player(); p.input.add(ShortArray(960) { 7 }); p.finished = true; p.device.limit = 480 + p.step(0); p.device.head = 240; p.step(5) + assertFalse(p.output.drained); assertEquals(480L, p.output.written) + p.device.head = 480; p.step(10) + assertTrue(p.output.drained); assertEquals(List(960) { 7.toShort() }, p.device.samples) + } + + @Test fun shortEofPrimesOldDeviceWithoutAdvancingMediaClock() { + val p = Player(Device(3848)); p.input.add(ShortArray(200) { 7 }); p.finished = true + repeat(12) { p.step(it * 5L) } + assertEquals(100L, p.output.written); assertEquals(0L, p.output.played) + assertEquals(200 + 3848 * 2, p.device.samples.size) + assertTrue(p.device.samples.drop(200).all { it == 0.toShort() }) + p.device.head = 200; p.step(65) + assertEquals(100L, p.output.played); assertTrue(p.output.drained) + } + + @Test fun eofAfterUnderrunAlsoPrimesOldDevice() { + val p = Player(Device(3848)); p.input.add(ShortArray(960)); p.step(0) + p.device.head = 480; p.step(5) + p.input.add(ShortArray(100) { 9 }); p.finished = true + repeat(12) { p.step(10 + it * 5L) } + assertEquals(530L, p.output.written) + assertTrue(p.device.samples.size / 2 - p.device.head >= p.device.threshold) + p.device.head = 530; p.step(75); assertTrue(p.output.drained) + } + + @Test fun resetDiscardsPendingAndNewOutputStartsAtZero() { + val p = Player(); p.input.add(ShortArray(960) { 8 }); p.device.limit = 120; p.step(0); p.output.close() + assertTrue(p.device.closed); assertEquals(0, p.output.pendingFrames) + val next = Player(); next.input.add(ShortArray(960) { 9 }); next.step(10) + assertEquals(480L, next.output.written); assertEquals(0L, next.output.played) + assertTrue(next.device.samples.all { it == 9.toShort() }) + } + + @Test fun writeErrorContainsDeviceAndQueueState() { + val p = Player(); p.input.add(ShortArray(960)); p.device.limit = -6 + val error = assertThrows(IllegalStateException::class.java) { p.step(0) } + assertTrue(error.message!!.contains("write=-6")); assertTrue(error.message!!.contains("pending=480")) + } + + @Test fun unexpectedHeadResetIsNotInterpretedAsFourBillionPlayedFrames() { + val p = Player(); p.input.add(ShortArray(960)); p.step(0); p.device.head = 100; p.step(5) + p.device.head = 0 + assertThrows(IllegalStateException::class.java) { p.step(10) } + } +}