Add tested adaptive playback device buffer policy

4 days ago · cbc7b296bb
2 changed files with 184 additions and 0 deletions
--- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallPlaybackBuffer.kt
+++ b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallPlaybackBuffer.kt
@ -0,0 +1,68 @@
+package com.utun.chat.data
+
+/** Запас устройства, не сетевой JB. Владелец сериализует sample/applied/settle одним lock. */
+internal class CallPlaybackBuffer(val blockFrames: Int = 960) {
+    data class Change(val frames: Int, val reason: String, val underruns: Int = 0)
+
+    private var actual = 0
+    private var capacity = 0
+    private var floor = blockFrames
+    private var previousUnderruns = 0
+    private var previousPlayed = -1L
+    private var previousMs = 0L
+    private var stableSinceMs = 0L
+    private var settleUntilMs = 0L
+    private var probeIntervalMs = 30_000L
+
+    fun reset(nowMs: Long, frames: Int, capacityFrames: Int, underruns: Int) {
+        require(frames > 0 && capacityFrames >= frames && underruns >= 0)
+        actual = frames
+        capacity = capacityFrames
+        floor = blockFrames
+        previousUnderruns = underruns
+        previousPlayed = -1
+        previousMs = nowMs
+        probeIntervalMs = 30_000
+        settle(nowMs)
+    }
+
+    /** Старт и запрос/подтверждение маршрута не участвуют в обучении. */
+    fun settle(nowMs: Long) {
+        settleUntilMs = nowMs + 2_000
+        stableSinceMs = settleUntilMs
+    }
+
+    fun sample(nowMs: Long, underruns: Int, played: Long): Change? {
+        val delta = underruns - previousUnderruns
+        val progress = played >= 0 && previousPlayed >= 0 && played > previousPlayed
+        val timely = nowMs - previousMs in 1..1_000
+        previousUnderruns = underruns
+        previousPlayed = played
+        previousMs = nowMs
+        if (nowMs < settleUntilMs) return null
+        if (delta < 0) {
+            settle(nowMs)
+            return Change(actual, "counter-reset")
+        }
+        if (delta > 0) {
+            probeIntervalMs = (probeIntervalMs * 2).coerceAtMost(300_000)
+            settle(nowMs)
+            val next = (actual + blockFrames).coerceAtMost(capacity)
+            if (next == actual) settleUntilMs = nowMs + 10_000
+            return Change(next, if (next > actual) "underrun" else "capacity-limit", delta)
+        }
+        if (!progress || !timely) stableSinceMs = nowMs
+        if (actual <= floor || nowMs - stableSinceMs < probeIntervalMs) return null
+        stableSinceMs = nowMs
+        return Change((actual - blockFrames).coerceAtLeast(floor), "stable-probe")
+    }
+
+    /** Устройство может округлить/ограничить заявку: учимся по принятому размеру. */
+    fun applied(nowMs: Long, change: Change, frames: Int, capacityFrames: Int) {
+        require(frames > 0 && capacityFrames >= frames)
+        if (change.frames < actual && frames >= actual) floor = frames
+        actual = frames
+        capacity = capacityFrames
+        settle(nowMs)
+    }
+}
--- a/tools/chatgui-android/app/src/test/java/com/utun/chat/data/CallPlaybackBufferTest.kt
+++ b/tools/chatgui-android/app/src/test/java/com/utun/chat/data/CallPlaybackBufferTest.kt
@ -0,0 +1,116 @@
+package com.utun.chat.data
+
+import org.junit.Assert.*
+import org.junit.Test
+
+class CallPlaybackBufferTest {
+    private class Device(val capacity: Int = 7680, val minimum: Int = 960) {
+        val policy = CallPlaybackBuffer()
+        var now = 0L
+        var played = 0L
+        var underruns = 0
+        var frames = 1920
+        val changes = mutableListOf<CallPlaybackBuffer.Change>()
+        init { policy.reset(now, frames, capacity, underruns) }
+        fun tick(advance: Boolean = true, duration: Long = 200): CallPlaybackBuffer.Change? {
+            now += duration
+            if (advance) played += duration * 48
+            val change = policy.sample(now, underruns, played) ?: return null
+            changes.add(change)
+            if (change.frames != frames) {
+                frames = change.frames.coerceIn(minimum, capacity)
+                policy.applied(now, change, frames, capacity)
+            }
+            return change
+        }
+        fun run(ms: Long, advance: Boolean = true) { repeat((ms / 200).toInt()) { tick(advance) } }
+    }
+
+    @Test fun stableDeviceReachesOneBlockAndStopsProbing() {
+        val d = Device()
+        d.run(31_800)
+        assertEquals(1920, d.frames)
+        assertEquals("stable-probe", d.tick()?.reason)
+        assertEquals(960, d.frames)
+        d.run(300_000)
+        assertEquals(1, d.changes.size)
+    }
+
+    @Test fun underrunGrowsImmediatelyAndDelaysNextProbe() {
+        val d = Device()
+        d.run(35_000)
+        d.underruns++
+        assertEquals("underrun", d.tick()?.reason)
+        assertEquals(1920, d.frames)
+        d.run(61_800)
+        assertEquals(1920, d.frames)
+        assertEquals("stable-probe", d.tick()?.reason)
+        assertEquals(960, d.frames)
+    }
+
+    @Test fun startupAndRouteUnderrunsDoNotTeachWrongSize() {
+        val d = Device()
+        d.underruns = 3
+        assertNull(d.tick())
+        d.run(20_000)
+        d.policy.settle(d.now) // Запрос маршрута ещё до framework callback.
+        d.underruns = 7
+        d.run(1_000, false)
+        d.policy.reset(d.now, d.frames, d.capacity, d.underruns) // Подтверждённый новый выход.
+        d.underruns = 9
+        d.run(31_800)
+        assertTrue(d.changes.isEmpty())
+        assertEquals("stable-probe", d.tick()?.reason)
+    }
+
+    @Test fun AcceptedHardwareMinimumIsNotProbedForever() {
+        val d = Device(minimum = 1440)
+        d.run(32_000)
+        assertEquals(1440, d.frames)
+        d.run(32_000)
+        assertEquals(1440, d.frames)
+        assertEquals(2, d.changes.size)
+        d.run(300_000)
+        assertEquals(2, d.changes.size)
+    }
+
+    @Test fun noPlaybackOrMissingObservationsCannotProveStability() {
+        val d = Device()
+        d.run(60_000, false)
+        assertTrue(d.changes.isEmpty())
+        d.run(20_000)
+        d.tick(duration = 40_000)
+        assertTrue(d.changes.isEmpty())
+        d.run(29_800)
+        assertEquals("stable-probe", d.tick()?.reason)
+    }
+
+    @Test fun realStarvationStillGrowsWhenHeadOrReporterStalls() {
+        val d = Device()
+        d.run(5_000)
+        d.underruns++
+        assertEquals("underrun", d.tick(false, 2_000)?.reason)
+        assertEquals(2880, d.frames)
+    }
+
+    @Test fun growthStopsAtCapacityAndRateLimitsWarnings() {
+        val d = Device(capacity = 1920)
+        d.run(5_000)
+        d.underruns++
+        assertEquals("capacity-limit", d.tick()?.reason)
+        repeat(49) { d.underruns++; assertNull(d.tick()) }
+        assertEquals(1, d.changes.size)
+        assertEquals(1920, d.frames)
+    }
+
+    @Test fun counterResetIsRebasedWithoutGrowingBuffer() {
+        val d = Device()
+        d.underruns = 10
+        d.run(5_000)
+        d.underruns = 0
+        assertEquals("counter-reset", d.tick()?.reason)
+        assertEquals(1920, d.frames)
+        d.run(32_000)
+        assertEquals(960, d.frames)
+    }
+}