From 5151c966321324561044081222395f654cf392e5 Mon Sep 17 00:00:00 2001 From: kiefer christian <8891890+MrKitty871@users.noreply.github.com> Date: Sun, 18 Oct 2026 04:31:26 -0500 Subject: [PATCH] fix(audio): preserve partial streaming PCM writes --- .../relay/audio/RealtimePcmPlayer.kt | 114 +++++++++++++----- .../audio/RealtimePcmBufferPolicyTest.kt | 80 ++++++++++++ 2 files changed, 167 insertions(+), 27 deletions(-) diff --git a/app/src/main/kotlin/com/hermesandroid/relay/audio/RealtimePcmPlayer.kt b/app/src/main/kotlin/com/hermesandroid/relay/audio/RealtimePcmPlayer.kt index dc8af06fd..0acaffd70 100644 --- a/app/src/main/kotlin/com/hermesandroid/relay/audio/RealtimePcmPlayer.kt +++ b/app/src/main/kotlin/com/hermesandroid/relay/audio/RealtimePcmPlayer.kt @@ -61,6 +61,7 @@ class RealtimePcmPlayer(private val context: Context? = null) { private var lastMutedVolumeDiagnosticAtMs: Long = 0L private var adaptiveStartPrebufferMs: Long = RealtimePcmBufferPolicy.START_PREBUFFER_MS private var playbackGapSeenThisTrack: Boolean = false + @Volatile private var stopEpoch: Long = 0L private val _amplitude = MutableStateFlow(0f) val amplitude: StateFlow = _amplitude.asStateFlow() @@ -74,18 +75,22 @@ class RealtimePcmPlayer(private val context: Context? = null) { if (pcm.isEmpty()) return 0f val level = computePcm16LeRms(pcm) val now = SystemClock.elapsedRealtime() + var publishEpoch = -1L + var publishTrack: AudioTrack? = null val written = synchronized(writeLock) { - val track = try { + val (track, writeEpoch) = try { synchronized(trackLock) { val currentTrack = ensureTrackLocked(sampleRate) notePlaybackGapLocked(currentTrack, now) - currentTrack + currentTrack to stopEpoch } } catch (e: Exception) { Log.w(TAG, "PCM track preparation failed: ${e.message}") synchronized(trackLock) { releaseTrackLocked(reason = "PCM track preparation failure") } return@synchronized 0 } + publishEpoch = writeEpoch + publishTrack = track try { val prerollWritten = maybeWriteStartupPreroll(track, sampleRate) @@ -103,7 +108,29 @@ class RealtimePcmPlayer(private val context: Context? = null) { // normal start decision, and the end-of-turn flush // (voice.output_audio.done) force-starts anything still buffered. - val writtenBytes = writeBlocking(track, pcm) + val writtenBytes = writePcmFully( + pcmSize = pcm.size, + writeChunk = { offset, size -> + writeBlocking(track, pcm, offset, size) + }, + onWritten = { acceptedBytes -> + synchronized(trackLock) { + if (audioTrack === track) { + noteWrittenBytesLocked(acceptedBytes, sampleRate) + enqueuePlaybackAmplitudeLocked(level) + // AudioTrack can accept only its fixed buffer capacity + // before play() starts, even for WRITE_BLOCKING. Start as + // soon as that first partial write reaches the prebuffer, + // then continue writing the remainder while playback drains + // the buffer instead of silently dropping the chunk tail. + maybeStartPlaybackLocked(track, sampleRate, force = false) + updateUnderrunCursorLocked(track) + logPlaybackHealthLocked(track, SystemClock.elapsedRealtime()) + } + } + }, + shouldContinue = { stopEpoch == writeEpoch }, + ) if (writtenBytes < 0) { Log.w(TAG, "PCM write returned $writtenBytes; restarting track") synchronized(trackLock) { @@ -112,20 +139,10 @@ class RealtimePcmPlayer(private val context: Context? = null) { return@synchronized 0 } - var accepted = 0 - if (writtenBytes > 0) { - synchronized(trackLock) { - if (audioTrack === track) { - noteWrittenBytesLocked(writtenBytes, sampleRate) - enqueuePlaybackAmplitudeLocked(level) - maybeStartPlaybackLocked(track, sampleRate, force = false) - updateUnderrunCursorLocked(track) - logPlaybackHealthLocked(track, now) - accepted = writtenBytes - } - } + if (writtenBytes in 0 until pcm.size) { + Log.w(TAG, "PCM write accepted only $writtenBytes/${pcm.size} bytes") } - accepted + writtenBytes } catch (e: Exception) { Log.w(TAG, "PCM write failed: ${e.message}") synchronized(trackLock) { @@ -134,18 +151,25 @@ class RealtimePcmPlayer(private val context: Context? = null) { return@synchronized 0 } } - if (written > 0) { - _amplitude.value = level + synchronized(trackLock) { + if (shouldPublishPcmAmplitude( + writtenBytes = written, + writeEpoch = publishEpoch, + currentEpoch = stopEpoch, + trackStillCurrent = audioTrack === publishTrack, + )) { + _amplitude.value = level + } } return level } - private fun writeBlocking(track: AudioTrack, pcm: ByteArray): Int = + private fun writeBlocking(track: AudioTrack, pcm: ByteArray, offset: Int = 0, size: Int = pcm.size): Int = if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.M) { - track.write(pcm, 0, pcm.size, AudioTrack.WRITE_BLOCKING) + track.write(pcm, offset, size, AudioTrack.WRITE_BLOCKING) } else { @Suppress("DEPRECATION") - track.write(pcm, 0, pcm.size) + track.write(pcm, offset, size) } fun flushBufferedPlayback(cushionMs: Long = DEFAULT_DRAIN_CUSHION_MS): Long { @@ -165,12 +189,14 @@ class RealtimePcmPlayer(private val context: Context? = null) { } fun stop() { - synchronized(writeLock) { - synchronized(trackLock) { - releaseTrackLocked(reason = "stop") - currentSampleRate = 0 - estimatedPlaybackEndAtMs = 0L - } + // Do not wait for writeLock: AudioTrack explicitly allows stop/release + // from another thread to interrupt a blocking write. Waiting here would + // make cancellation take as long as the remaining provider chunk. + synchronized(trackLock) { + stopEpoch++ + releaseTrackLocked(reason = "stop") + currentSampleRate = 0 + estimatedPlaybackEndAtMs = 0L } _amplitude.value = 0f } @@ -749,6 +775,40 @@ internal fun amplitudeAtHead(queue: List, headFrames: Long): Float { return 0f } +/** + * Drains one logical PCM chunk through AudioTrack's partial-write contract. + * + * A stopped MODE_STREAM AudioTrack may accept only its hardware buffer capacity + * even when WRITE_BLOCKING is requested. The callback lets the caller start + * playback after that first accepted prefix, allowing subsequent writes to drain + * the rest of the same provider chunk without losing its tail. + */ +internal inline fun writePcmFully( + pcmSize: Int, + writeChunk: (offset: Int, size: Int) -> Int, + onWritten: (writtenBytes: Int) -> Unit, + shouldContinue: () -> Boolean = { true }, +): Int { + var offset = 0 + while (offset < pcmSize && shouldContinue()) { + val written = writeChunk(offset, pcmSize - offset) + if (written < 0) return written + if (written == 0) break + val accepted = written.coerceAtMost(pcmSize - offset) + offset += accepted + onWritten(accepted) + } + return offset +} + +/** Prevents an interrupted writer from reviving UI amplitude after stop(). */ +internal fun shouldPublishPcmAmplitude( + writtenBytes: Int, + writeEpoch: Long, + currentEpoch: Long, + trackStillCurrent: Boolean, +): Boolean = writtenBytes > 0 && writeEpoch == currentEpoch && trackStillCurrent + internal data class RealtimePcmStartDecision( val shouldStart: Boolean, val bufferedMs: Long, diff --git a/app/src/test/kotlin/com/hermesandroid/relay/audio/RealtimePcmBufferPolicyTest.kt b/app/src/test/kotlin/com/hermesandroid/relay/audio/RealtimePcmBufferPolicyTest.kt index 1215417b0..b55f11348 100644 --- a/app/src/test/kotlin/com/hermesandroid/relay/audio/RealtimePcmBufferPolicyTest.kt +++ b/app/src/test/kotlin/com/hermesandroid/relay/audio/RealtimePcmBufferPolicyTest.kt @@ -6,6 +6,86 @@ import org.junit.Assert.assertTrue import org.junit.Test class RealtimePcmBufferPolicyTest { + @Test + fun `partial stopped-track write preserves the complete provider chunk`() { + val writes = mutableListOf>() + val accepted = mutableListOf() + var call = 0 + + val total = writePcmFully( + pcmSize = 192_800, + writeChunk = { offset, size -> + writes += offset to size + if (call++ == 0) 33_600 else size + }, + onWritten = accepted::add, + ) + + assertEquals(192_800, total) + assertEquals(listOf(0 to 192_800, 33_600 to 159_200), writes) + assertEquals(listOf(33_600, 159_200), accepted) + } + + @Test + fun `zero-progress write stops without spinning`() { + var calls = 0 + + val total = writePcmFully( + pcmSize = 48_000, + writeChunk = { _, _ -> + calls++ + 0 + }, + onWritten = { error("zero-byte writes must not be reported as accepted") }, + ) + + assertEquals(0, total) + assertEquals(1, calls) + } + + @Test + fun `stop signal cancels remaining partial writes`() { + val writes = mutableListOf>() + var active = true + + val total = writePcmFully( + pcmSize = 192_800, + writeChunk = { offset, size -> + writes += offset to size + 33_600 + }, + onWritten = { active = false }, + shouldContinue = { active }, + ) + + assertEquals(33_600, total) + assertEquals(listOf(0 to 192_800), writes) + } + + @Test + fun `interrupted write cannot revive amplitude after stop`() { + assertFalse( + shouldPublishPcmAmplitude( + writtenBytes = 33_600, + writeEpoch = 4L, + currentEpoch = 5L, + trackStillCurrent = false, + ), + ) + } + + @Test + fun `completed current write publishes amplitude`() { + assertTrue( + shouldPublishPcmAmplitude( + writtenBytes = 192_800, + writeEpoch = 5L, + currentEpoch = 5L, + trackStillCurrent = true, + ), + ) + } + @Test fun `does not start on tiny first realtime chunk`() { val decision = RealtimePcmBufferPolicy.startDecision(