Fix Android stream audio backpressure without blocking video

2026-10-14 17:16:45 +02:00
1 parent 91d1a7994e
commit 0fdc6e435b
@@ -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()
+    }
+}
@@ -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<Frame>
        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