Music reaches a car that acks slowly in chunks as long as its gap

The 2020 Civic (issue #43) acks one audio chunk every 45 to 85 ms whatever it holds, so the
21 ms chunks Gearslip sends reached it at a quarter to half of real time: a second of music,
then silence. While the window is full, AudioLink now measures the car's gap between acks, and
when it's well over what a chunk holds, gathers the capture into chunks that last the gap,
rounded up to a power of two in frames. The Civic gets 4096 frames. A car that keeps up never
fills the window and gets the capture as it comes.

gearslip/src/main/java/app/seb3thehacker/gearslip/audio/AudioLink.kt

@@ -52,6 +52,21 @@ class AudioLink(
     // The Uconnect asks for 4 on its video channel; assume the same until its Config says otherwise.
     @Volatile private var maxUnacked = 4
     @Volatile private var lastAckAt = 0L
+    /**
+     * Bytes of PCM per Data message, or 0 to send each capture read as it comes. Raised by
+     * [adaptChunkSize] for a car that acks fewer chunks than it plays.
+     */
+    @Volatile private var chunkBytes = 0
+    /** Capture reads gathered until they fill [chunkBytes]. Only the capture thread touches it. */
+    private val pending = java.io.ByteArrayOutputStream()
+    /** Set by a new stream, so the capture thread drops what an older one left in [pending]. */
+    @Volatile private var pendingStale = false
+    /** The size of the last chunk sent, for [adaptChunkSize] to know what each one holds. */
+    @Volatile private var lastPayloadBytes = 0
+    /** Acks that came while the window was full, and the time they took: the car's own pace. */
+    private var slowAcks = 0
+    private var slowAckMs = 0L
+    private var lastAckFull = false
     private var stalls = 0
     private var chunks = 0
     private var bytes = 0L
@@ -211,11 +226,21 @@ class AudioLink(
 
             MSG_ACK -> {
                 val acked = Wire.varint(Wire.fields(body), 2)?.toInt() ?: 1
+                val now = System.currentTimeMillis()
                 synchronized(ackLock) {
+                    // A full window means the sender is waiting on this ack. When it was waiting on
+                    // the one before too, the gap between them is the car's own pace.
+                    val full = inFlight >= maxUnacked
+                    if (full && lastAckFull) {
+                        slowAcks++
+                        slowAckMs += now - lastAckAt
+                    }
+                    lastAckFull = full
                     inFlight = maxOf(0, inFlight - maxOf(1, acked))
-                    lastAckAt = System.currentTimeMillis()
+                    lastAckAt = now
                     ackLock.notifyAll()
                 }
+                adaptChunkSize()
             }
 
             // Whatever else the head unit answers Setup with, we need its bytes to identify it.
@@ -290,6 +315,40 @@ class AudioLink(
         }
     }
 
+    /**
+     * A 2020 Civic acks a chunk about every 83 ms whatever the chunk holds, so 21 ms chunks
+     * reached it at a quarter of real time: a second of music, then silence, over and over
+     * (issue #43). When the car keeps the sender waiting and its acks come much further apart
+     * than the audio in each chunk, gather the capture into chunks that last at least that gap,
+     * rounded up to a power of two in frames: 83 ms is about 4096 frames at 48 kHz, which looks
+     * like the car's own buffer. A car that keeps up never fills the window, so it never gets here.
+     */
+    private fun adaptChunkSize() {
+        if (aac != null) return
+        val interval = synchronized(ackLock) {
+            if (slowAcks < ADAPT_ACKS) return
+            (slowAckMs.toDouble() / slowAcks).also {
+                slowAcks = 0
+                slowAckMs = 0
+            }
+        }
+        val frame = channels * 2
+        val bytesPerMs = sampleRate * frame / 1000.0
+        val current = if (chunkBytes > 0) chunkBytes else lastPayloadBytes
+        if (current <= 0 || current >= MAX_CHUNK_BYTES) return
+        val heldMs = current / bytesPerMs
+        if (interval < heldMs * SLOW_ACK_RATIO) return
+        val gapFrames = Math.ceil(interval * sampleRate / 1000.0).toInt().coerceAtLeast(1)
+        val frames = Integer.highestOneBit(gapFrames).let { if (it < gapFrames) it shl 1 else it }
+        val target = minOf(MAX_CHUNK_BYTES / frame * frame, frames * frame)
+        if (target <= current) return
+        chunkBytes = target
+        GearslipLog.i(
+            "$tag: the car acks a chunk every ${interval.toInt()}ms, but each holds ${heldMs.toInt()}ms - " +
+                "sending $target-byte chunks (${(target / bytesPerMs).toInt()}ms) from now on",
+        )
+    }
+
     /** Silence for [ms], whole frames of this sink's format. */
     private fun silence(ms: Int): ByteArray = ByteArray(sampleRate * ms / 1000 * channels * 2)
 
@@ -331,6 +390,12 @@ class AudioLink(
         }
         inFlight = 0
         lastAckAt = System.currentTimeMillis()
+        pendingStale = true
+        synchronized(ackLock) {
+            slowAcks = 0
+            slowAckMs = 0
+            lastAckFull = false
+        }
         if (aacSink && aac == null) {
             aac = runCatching { AacEncoder(sampleRate, channels, adts = service.codecType == ServiceDiscovery.CODEC_AAC_LC_ADTS) }
                 .onSuccess { GearslipLog.i("$tag: the car takes ${service.codecName} only - encoding to AAC") }
@@ -380,7 +445,20 @@ class AudioLink(
     private fun writeNow(pcm: ByteArray, size: Int) {
         val encoder = aac
         if (encoder == null) {
-            send(pcm.copyOf(size))
+            if (pendingStale) {
+                pendingStale = false
+                pending.reset()
+            }
+            val want = chunkBytes
+            if (want <= 0 && pending.size() == 0) {
+                send(pcm.copyOf(size))
+                return
+            }
+            pending.write(pcm, 0, size)
+            if (pending.size() >= want) {
+                send(pending.toByteArray())
+                pending.reset()
+            }
             return
         }
         val frames = try {
@@ -430,6 +508,7 @@ class AudioLink(
             return
         }
         synchronized(ackLock) { inFlight++ }
+        lastPayloadBytes = payload.size
         chunks++
         bytes += payload.size
         // The first chunk proves the whole path; after that, an occasional line is enough to show
@@ -478,6 +557,12 @@ class AudioLink(
         private const val FOCUS_STATE_LOSS_TRANSIENT = 5
         private const val SESSION_ID = 2
         private const val STALL_MS = 2_000L
+        /** Acks measured with the window full before the chunk size is judged. */
+        private const val ADAPT_ACKS = 20
+        /** Acks this much further apart than a chunk lasts mean the car is the one holding back. */
+        private const val SLOW_ACK_RATIO = 1.5
+        /** About 170 ms at 48 kHz stereo: the most delay a bigger chunk may add. */
+        private const val MAX_CHUNK_BYTES = 32_768
         private const val CHUNK_LOG_EVERY = 200 // about every four seconds
         private const val CONFIG_WAIT_MS = 1_000L
 

gearslip/src/test/java/app/seb3thehacker/gearslip/audio/AudioLinkChunkTest.kt

@@ -0,0 +1,94 @@
+package app.seb3thehacker.gearslip.audio
+
+import app.seb3thehacker.gearslip.Protobuf
+import app.seb3thehacker.gearslip.ServiceDiscovery
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Test
+import java.util.Collections
+
+/**
+ * A 2020 Civic acks a music chunk about every 83 ms, whatever the chunk holds (issue #43).
+ * Capture reads of 4096 bytes hold 21 ms at 48 kHz stereo, so the music reached the car at a
+ * quarter of real time. [AudioLink] should notice the car's pace and send chunks that last as
+ * long as its gap between acks; a car that keeps up should get the capture as it comes.
+ */
+class AudioLinkChunkTest {
+
+    private val sink = ServiceDiscovery.AudioService(
+        serviceId = 4,
+        streamType = 3,
+        codecType = ServiceDiscovery.CODEC_PCM,
+        configs = listOf(ServiceDiscovery.AudioConfig(48_000, 16, 2)),
+    )
+
+    /** A started link to a car that takes [window] chunks before it acks, and what it was sent. */
+    private fun startedLink(window: Int): Pair<AudioLink, MutableList<Int>> {
+        val sent = Collections.synchronizedList(mutableListOf<Int>())
+        val link = AudioLink(
+            sink,
+            sendOnChannel = { id, body -> if (id == MSG_DATA) sent += body.size - TIMESTAMP_BYTES },
+            sendControl = { _, _ -> },
+            asksFocus = false,
+            onReady = {},
+            onClose = {},
+        )
+        // Config { status = READY; max_unacked = window; configuration_indices = 0 }
+        link.onMessage(
+            MSG_CONFIG,
+            Protobuf.varintField(1, 2) + Protobuf.varintField(2, window.toLong()) + Protobuf.varintField(3, 0),
+        )
+        link.start()
+        return link to sent
+    }
+
+    private fun ack(link: AudioLink) =
+        link.onMessage(MSG_ACK, Protobuf.varintField(1, 2) + Protobuf.varintField(2, 1))
+
+    @Test
+    fun aCarThatAcksSlowlyGetsChunksAsLongAsItsGap() {
+        val (link, sent) = startedLink(window = 16)
+        val capture = Thread {
+            val read = ByteArray(CAPTURE_BYTES)
+            while (!Thread.currentThread().isInterrupted) link.write(read, read.size)
+        }.apply { isDaemon = true; start() }
+        try {
+            // The car acks one chunk every 60 ms. The capture thread fills the window at once
+            // and then waits on each ack, so each gap is the car's own pace.
+            Thread.sleep(200)
+            repeat(30) {
+                Thread.sleep(60)
+                ack(link)
+            }
+            Thread.sleep(100)
+        } finally {
+            capture.interrupt()
+            link.stop()
+        }
+        // 60 ms is 2880 frames at 48 kHz, rounded up to 4096 frames of 4 bytes.
+        assertTrue("first chunks go out as captured: $sent", sent.take(16).all { it == CAPTURE_BYTES })
+        assertEquals(16_384, sent.last())
+    }
+
+    @Test
+    fun aCarThatKeepsUpGetsTheCaptureAsItComes() {
+        val (link, sent) = startedLink(window = 16)
+        val read = ByteArray(CAPTURE_BYTES)
+        // Every chunk is acked before the next, so the window never fills.
+        repeat(100) {
+            link.write(read, read.size)
+            ack(link)
+        }
+        link.stop()
+        assertEquals(100, sent.size)
+        assertTrue(sent.all { it == CAPTURE_BYTES })
+    }
+
+    private companion object {
+        const val MSG_DATA = 0
+        const val MSG_CONFIG = 32771
+        const val MSG_ACK = 32772
+        const val TIMESTAMP_BYTES = 8
+        const val CAPTURE_BYTES = 4096
+    }
+}