Fix Opus decoding and RX jitter buffer on Android 16 (v1.0.5)

2026-08-17 09:22:47 +03:00
parent 268c9d76af
commit 0b51e63262
2 changed files with 183 additions and 71 deletions
--- a/app/src/main/java/ru/ua1zbe/svxremote/audio/OpusEngine.kt
+++ b/app/src/main/java/ru/ua1zbe/svxremote/audio/OpusEngine.kt
@@ -47,33 +47,39 @@ class OpusEngine(
    private var recThread: Thread? = null

    private val encodeBufferInfo = MediaCodec.BufferInfo()
-    private val decodeBufferInfo = MediaCodec.BufferInfo()

    // ---------- Воспроизведение (RX) ----------

-    private var headSent = false
    private val rxQueue = java.util.concurrent.ConcurrentLinkedQueue<ByteArray>()
-    private var drainThread: Thread? = null
+    private val freeInputs = java.util.concurrent.ConcurrentLinkedQueue<Int>()
    private var ptsCounter = 0L
+    private var receivedAny = false
+    private var concealLogs = 0
+    private var dbgOut = 0
+    private var playStarted = false
+    private var gateStart = 0L
+    private val jitterTarget = 6 // ≈ 120 мс накапливаем до первого слива
+    private val configBuffers = ArrayList<ByteArray>()
+    private var configIdx = 0

    fun startDecoder() {
        if (decoder != null) return
        val format = MediaFormat()
        format.setString(MediaFormat.KEY_MIME, MediaFormat.MIMETYPE_AUDIO_OPUS)
-        format.setInteger(MediaFormat.KEY_SAMPLE_RATE, SAMPLE_RATE)
+        // ВАЖНО: c2.android.opus.decoder (Android 16) выводит только 48 кГц
+        // (kRate=48000); KEY_SAMPLE_RATE=8000 вызывает конфликт конфига
+        // («We have a failed config») и фатальную ошибку кодера.
+        format.setInteger(MediaFormat.KEY_SAMPLE_RATE, OUT_SAMPLE_RATE)
        format.setInteger(MediaFormat.KEY_CHANNEL_COUNT, CHANNELS)
-        format.setInteger(MediaFormat.KEY_MAX_INPUT_SIZE, 8192)
-        // Android 16: c2.android.opus.decoder ждёт 3 конфиг-буфера:
-        // csd-0 OpusHead, csd-1 CodecDelay (int64 LE нс), csd-2 SeekPreRoll (int64 LE нс).
-        // Без csd-1/2 декодер берёт мусорные значения из пакетов и молча
-        // выкидывает весь вывод (mSamplesToDiscard).
-        format.setByteBuffer("csd-0", ByteBuffer.wrap(opusHead()))
-        // Прескип 312 сэмплов при 8 кГц = 39 мс = 39_000_000 нс (перекодируется в 48 кГц внутри)
-        format.setByteBuffer("csd-1", ByteBuffer.wrap(longBytes(39_000_000L)))
-        format.setByteBuffer("csd-2", ByteBuffer.wrap(longBytes(0L)))
+        // Android 16: c2.android.opus.decoder читает конфиг (OpusHead + codec
+        // delay + seek preroll) ТОЛЬКО из первых входных буферов
+        // (см. queueDecoderConfig); csd-ключи MediaFormat он игнорирует, и без
+        // конфига первый же аудио-кадр ломает декодер (C2_CORRUPTED).
        decoder = try {
            MediaCodec.createDecoderByType(MediaFormat.MIMETYPE_AUDIO_OPUS).also {
+                it.setCallback(decoderCallback)
                it.configure(format, null, null, 0)
+                queueDecoderConfig() // готовим конфиг ДО start: колбэки уйдут после
                it.start()
            }
        } catch (e: Exception) {
@@ -81,6 +87,7 @@ class OpusEngine(
            null
        }
        if (decoder == null) return
+        Log.d(TAG, "decoder started, config=${configBuffers.size}B")
        val minBuf = AudioTrack.getMinBufferSize(
            OUT_SAMPLE_RATE,
            AudioFormat.CHANNEL_OUT_MONO,
@@ -92,77 +99,165 @@ class OpusEngine(
                OUT_SAMPLE_RATE,
                AudioFormat.CHANNEL_OUT_MONO,
                AudioFormat.ENCODING_PCM_16BIT,
-                maxOf(minBuf, FRAME_SIZE * 2 * 8),
+                maxOf(minBuf, 20000), // ~208 мс — буфер AudioTrack и есть джиттер-буфер
                AudioTrack.MODE_STREAM
-            ).also { it.play() }
+            ) // play() включаем на первом выходном кадре — буфер успевает накопиться
        } catch (e: Exception) {
            Log.e(TAG, "AudioTrack create failed: ${e.message}")
            null
        }
-        headSent = true // csd уходит через формат, не через буферы
-        drainThread = Thread({ drainLoop() }, "opus-out").also { it.start() }
+        concealLogs = 0
+        dbgOut = 0
+        ptsCounter = 20_000L // первый кадр: pts!=0, иначе декодер повторно сбросит codecDelay
        Log.d(TAG, "decoder started")
    }

+    /** Асинхронные колбэки кодера: подача входных кадров + слив вывода. */
+    private val decoderCallback = object : MediaCodec.Callback() {
+        override fun onError(codec: MediaCodec, e: MediaCodec.CodecException) {
+            Log.e(TAG, "decoder error: ${e.message} code=${e.errorCode} transient=${e.isTransient}")
+        }
+
+        override fun onInputBufferAvailable(codec: MediaCodec, index: Int) {
+            // Конфиг (csd) первыми входными буферами — в async-режиме нельзя
+            // использовать dequeueInputBuffer, только колбэки
+            if (configIdx < configBuffers.size) {
+                val data = configBuffers[configIdx++]
+                try {
+                    val buf = codec.getInputBuffer(index) ?: return
+                    buf.clear()
+                    buf.put(data)
+                    codec.queueInputBuffer(index, 0, data.size, 0, MediaCodec.BUFFER_FLAG_CODEC_CONFIG)
+                    Log.d(TAG, "cfg ${data.size}B")
+                } catch (e: Exception) {
+                    Log.e(TAG, "cfg error: ${e.message}")
+                }
+                return
+            }
+            val pkt = rxQueue.poll()
+            when {
+                // Джиттер-гейт: пока не накопили jitterTarget кадров (или не
+                // вышло 80 мс с первого пакета) — не кормим декодер, чтобы
+                // буфер AudioTrack успел заполниться до старта воспроизведения
+                pkt != null && gateOpen() -> feedDecoder(codec, index, pkt)
+                pkt != null -> {
+                    rxQueue.add(pkt)
+                    freeInputs.add(index)
+                }
+                receivedAny -> {
+                    if (concealLogs < 3) {
+                        Log.d(TAG, "conceal (no packet)")
+                        concealLogs++
+                    }
+                    feedDecoder(codec, index, silenceFrame()) // скрытие потерь
+                }
+                else -> freeInputs.add(index) // ждём первый пакет
+            }
+        }
+
+        override fun onOutputBufferAvailable(codec: MediaCodec, index: Int, info: MediaCodec.BufferInfo) {
+            try {
+                if (info.size > 0) {
+                    val outBuf = ByteBuffer.allocate(info.size)
+                    outBuf.order(ByteOrder.LITTLE_ENDIAN)
+                    codec.getOutputBuffer(index)?.also { src ->
+                        src.position(info.offset)
+                        src.limit(info.offset + info.size)
+                        outBuf.put(src)
+                    }
+                    val pcm = outBuf.array()
+                    if (!playStarted) {
+                        playStarted = true
+                        Log.d(TAG, "play start")
+                        try {
+                            player?.play()
+                        } catch (e: Exception) {
+                            Log.e(TAG, "play error: ${e.message}")
+                        }
+                    }
+                    player?.write(pcm, 0, pcm.size)
+                    onLevel(SRC_SPEAKER, pcmLevel(pcm))
+                    if (dbgOut < 5) {
+                        Log.d(TAG, "out ${pcm.size}B")
+                        dbgOut++
+                    }
+                }
+            } catch (e: Exception) {
+                Log.e(TAG, "out error: ${e.message}")
+            } finally {
+                try {
+                    codec.releaseOutputBuffer(index, false)
+                } catch (_: Exception) {
+                }
+            }
+        }
+
+        override fun onOutputFormatChanged(codec: MediaCodec, format: MediaFormat) {
+            Log.d(TAG, "dec out format: sr=${format.getInteger(MediaFormat.KEY_SAMPLE_RATE)} " +
+                "ch=${format.getInteger(MediaFormat.KEY_CHANNEL_COUNT)}")
+        }
+    }
+
+    private fun feedDecoder(codec: MediaCodec, index: Int, pkt: ByteArray) {
+        try {
+            val buf = codec.getInputBuffer(index) ?: return
+            buf.clear()
+            buf.put(pkt)
+            codec.queueInputBuffer(index, 0, pkt.size, ptsCounter, 0)
+            ptsCounter += 20_000L // 20 мс кадра
+            if (concealLogs < 3 && rxQueue.isEmpty() && pkt === silenceFrame()) {
+                Log.d(TAG, "conceal (no packet)")
+                concealLogs++
+            }
+        } catch (e: Exception) {
+            Log.e(TAG, "feed error: ${e.message}")
+        }
+    }
+
+    /**
+     * Android 16: c2.android.opus.decoder читает конфиг из первых входных
+     * буферов. Единый CSD-блоб (формат AOSP WriteOpusHeaders): OpusHead (19
+     * байт) + codecDelay (int64 LE нс) + seekPreroll (int64 LE нс) — БЕЗ
+     * u32-префиксов длины. После него декодер уже сконфигурирован; лишние
+     * конфиг-буферы игнорируются как повторный CSD, поэтому шлём и их.
+     */
+    private fun queueDecoderConfig() {
+        configBuffers.clear()
+        try {
+            val head = opusHead()
+            val delay = longBytes(39_000_000L) // прескип 312 сэмплов при 8 кГц = 39 мс
+            val preroll = longBytes(0L)
+            val w = java.io.ByteArrayOutputStream()
+            w.write(head)
+            w.write(delay)
+            w.write(preroll)
+            configBuffers.add(w.toByteArray())
+            configBuffers.add(delay)
+            configBuffers.add(preroll)
+            configIdx = 0
+        } catch (_: Exception) {
+        }
+    }
+
    /** int64 в little-endian. */
    private fun longBytes(v: Long): ByteArray =
        ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN).putLong(v).array()

-    /** Отдельный поток: голова + входная очередь + слив вывода. */
-    private fun drainLoop() {
-        val dec = decoder ?: return
-        var dbg = 0
-        while (decoder != null && decoder === dec) {
-            try {
-                // Монотонный pts обязателен: при pts==0 c2.opus.decoder
-                // считает каждый кадр «перемоткой в ноль» и выбрасывает
-                // mCodecDelay сэмплов (C2SoftOpusDec.cpp, "When seeking to zero").
-                var pts = ptsCounter
-                while (true) {
-                    val pkt = rxQueue.poll() ?: break
-                    val inIdx = dec.dequeueInputBuffer(5000)
-                    if (inIdx < 0) { rxQueue.add(pkt); break }
-                    val buf = dec.getInputBuffer(inIdx) ?: break
-                    buf.clear()
-                    buf.put(pkt)
-                    dec.queueInputBuffer(inIdx, 0, pkt.size, pts, 0)
-                    pts += 20_000L // 20 мс кадра
-                    ptsCounter = pts
-                    if (dbg < 5) { Log.d(TAG, "queued ${pkt.size}B"); dbg++ }
-                }
-                var outIdx = dec.dequeueOutputBuffer(decodeBufferInfo, 1000)
-                while (outIdx >= 0) {
-                    if (outIdx == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED) {
-                        val f = dec.outputFormat
-                        Log.d(TAG, "dec out format: sr=${f.getInteger(MediaFormat.KEY_SAMPLE_RATE)} " +
-                            "ch=${f.getInteger(MediaFormat.KEY_CHANNEL_COUNT)}")
-                    } else {
-                        if (decodeBufferInfo.size > 0) {
-                            val outBuf = ByteBuffer.allocate(decodeBufferInfo.size)
-                            outBuf.order(ByteOrder.LITTLE_ENDIAN)
-                            dec.getOutputBuffer(outIdx)?.also { src ->
-                                src.position(decodeBufferInfo.offset)
-                                src.limit(decodeBufferInfo.offset + decodeBufferInfo.size)
-                                outBuf.put(src)
-                            }
-                            val pcm = outBuf.array()
-                            player?.write(pcm, 0, pcm.size)
-                            onLevel(SRC_SPEAKER, pcmLevel(pcm))
-                            if (dbg < 5) { Log.d(TAG, "out ${pcm.size}B"); dbg++ }
-                        }
-                        dec.releaseOutputBuffer(outIdx, false)
-                    }
-                    outIdx = dec.dequeueOutputBuffer(decodeBufferInfo, 0)
-                }
-            } catch (e: Exception) {
-                Log.e(TAG, "drain error: ${e.message}")
-            }
-        }
-    }
-
-    /** Подача одного Opus-пакета на декодер. */
+    /** Подача одного Opus-пакета на декодер (с UDP-потока). */
    fun onAudioPacket(packet: ByteArray) {
        rxQueue.add(packet)
+        receivedAny = true
+        if (gateStart == 0L) gateStart = System.currentTimeMillis()
+        if (!gateOpen()) return
+        val idx = freeInputs.poll() ?: return // если слот свободен — кормим сразу
+        decoder?.let { feedDecoder(it, idx, rxQueue.poll() ?: packet) }
+    }
+
+    /** Джиттер-гейт открыт: накопили кадры или вышло 80 мс с первого пакета. */
+    private fun gateOpen(): Boolean {
+        if (playStarted) return true
+        if (rxQueue.size >= jitterTarget) return true
+        return gateStart != 0L && System.currentTimeMillis() - gateStart >= 80
    }

    fun onAudioFlush() {
@@ -170,11 +265,23 @@ class OpusEngine(
            decoder?.flush()
            player?.flush()
            rxQueue.clear()
+            receivedAny = false
+            playStarted = false
+            gateStart = 0L
            onLevel(SRC_SPEAKER, 0f)
        } catch (_: Exception) {
        }
    }

+    /**
+     * Один кадр тишины (20 мс) для скрытия потерь вместо паузы в звуке.
+     * Сгенерирован opuslib: 8 кГц, моно, 20 кбит/с, нули PCM.
+     */
+    private fun silenceFrame(): ByteArray = byteArrayOf(
+        0x08.toByte(), 0x0B.toByte(), 0xE4.toByte(), 0xB9.toByte(),
+        0xA0.toByte(), 0xBC.toByte(), 0x84.toByte()
+    )
+
    /** Заголовок OpusHead (csd-0) для декодера: Opus, 1 канал, 8 кГц, прескип 312. */
    private fun opusHead(): ByteArray {
        val head = "OpusHead".toByteArray(Charsets.US_ASCII)
@@ -197,8 +304,12 @@ class OpusEngine(
    /** Остановить декодер и плеер (при отключении). */
    fun stopDecoder() {
        val dec = decoder
-        decoder = null // поток разбора выйдет из цикла до release
+        decoder = null
        rxQueue.clear()
+        freeInputs.clear()
+        receivedAny = false
+        playStarted = false
+        gateStart = 0L
        if (dec != null) {
            try {
                dec.stop()
--- a/app/src/main/java/ru/ua1zbe/svxremote/proto/ReflectorClient.kt
+++ b/app/src/main/java/ru/ua1zbe/svxremote/proto/ReflectorClient.kt
@@ -474,6 +474,7 @@ class ReflectorClient(
            stopUdp()
            val ds = DatagramSocket()
            udpSocket = ds
+            Log.d(TAG, "UDP bound port=${ds.localPort}")
            // Первый пакет регистрирует наш UDP-адрес у сервера
            val addr = resolvedAddr ?: InetAddress.getByName(s.host)
            sendRawUdp(ds, addr, s.port, MsgBuilder.udpHeartbeat(clientId, nextSeq()))