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) }
+ }
+}