feat(webrtc): 新增解码耗时统计并修复黑屏问题

1. 新增自编码与全托管模式的解码耗时探针（VideoSink），统计帧率与解码耗时。 2. 修复 SPS 与 PPS 作为独立 CONFIG 单元发送时被误丢弃导致 H.264 永久黑屏的问题。 3. 优化分片重组缓冲，丢弃过期单元并限制缓冲堆积以防内存泄漏。
2026-07-25 01:50:00 +08:00
parent cebb1e484e
commit 4a33f3db0d
3 changed files with 408 additions and 13 deletions
--- a/webrtc_controller_flutter/android/app/src/main/kotlin/com/ttstd/fluttercontroller/SelfCodecDecoderPlugin.kt
+++ b/webrtc_controller_flutter/android/app/src/main/kotlin/com/ttstd/fluttercontroller/SelfCodecDecoderPlugin.kt
@@ -16,6 +16,10 @@ import io.flutter.view.TextureRegistry
 import java.io.ByteArrayOutputStream
 import java.nio.ByteBuffer
 import java.nio.ByteOrder
+import org.webrtc.VideoFrame
+import org.webrtc.VideoSink
+import org.webrtc.VideoTrack
+import com.cloudwebrtc.webrtc.FlutterWebRTCPlugin

 private const val CHANNEL = "com.ttstd.fluttercontroller/self_codec"
 private const val TAG = "SelfCodecDecoderPlugin"
@@ -27,6 +31,8 @@ private const val UNIT_TYPE_CONFIG = 1
 private const val UNIT_TYPE_FRAME = 2
 private const val PLACEHOLDER_W = 1920
 private const val PLACEHOLDER_H = 1080
+// 解码耗时滑动平均窗口（帧数）。
+private const val DECODE_STATS_WINDOW = 90

 /// 单个单元的分片重组缓冲。
 private class ChunkBuffer(val seq: Int, val total: Int) {
@@ -52,6 +58,80 @@ private class ChunkBuffer(val seq: Int, val total: Int) {

 private class Frame(val data: ByteArray, val pts: Long, val key: Boolean)

+/// WebRTC 全托管模式下的解码耗时探针：给远端视频轨道额外挂一个 [VideoSink]，
+/// 逐帧记录真实到达时间，计算帧间耗时（≈真实解码+渲染周期）的 last/avg/peak 与帧率。
+/// 与自编码解码器不同，这里拿不到 MediaCodec 的单帧解码开销，只能测"连续两帧被解码
+/// 交付"的真实时间间隔——它是原生层实测值，不会像 totalDecodeTime 那样恒为 0。
+private class DecodeProbeSink {
+    private val lock = Any()
+    private var lastArrivalNs: Long = 0
+    private var sumMs = 0.0
+    private var count = 0
+    private var lastMs = 0.0
+    private var avgMs = 0.0
+    private var peakMs = 0.0
+    private var lastFpsStampMs = 0L
+    private var framesSinceFps = 0
+    private var fpsNow = 0
+
+    fun onFrame() {
+        val now = System.nanoTime()
+        synchronized(lock) {
+            if (lastArrivalNs != 0L) {
+                val deltaMs = (now - lastArrivalNs) / 1_000_000.0
+                // 丢弃异常跳变（首帧 / 长时间卡顿后的大间隔），避免污染统计。
+                if (deltaMs in 0.5..1000.0) {
+                    lastMs = deltaMs
+                    if (peakMs < deltaMs) peakMs = deltaMs
+                    count++
+                    sumMs += deltaMs
+                    avgMs = sumMs / count
+                    // 1 秒滑动窗口估算当前帧率。
+                    val nowMs = System.currentTimeMillis()
+                    framesSinceFps++
+                    if (lastFpsStampMs != 0L && nowMs - lastFpsStampMs >= 1000) {
+                        fpsNow = framesSinceFps * 1000 / (nowMs - lastFpsStampMs)
+                        framesSinceFps = 0
+                        lastFpsStampMs = nowMs
+                    } else if (lastFpsStampMs == 0L) {
+                        lastFpsStampMs = nowMs
+                    }
+                }
+            }
+            lastArrivalNs = now
+        }
+    }
+
+    fun snapshot(): Map<String, Any> = synchronized(lock) {
+        mapOf(
+            "lastMs" to lastMs,
+            "avgMs" to avgMs,
+            "peakMs" to peakMs,
+            "totalFrames" to count,
+            "fpsNow" to fpsNow,
+            "fpsAvg" to if (avgMs > 0) (1000.0 / avgMs).toInt() else 0
+        )
+    }
+
+    fun reset() = synchronized(lock) {
+        lastArrivalNs = 0
+        sumMs = 0.0
+        count = 0
+        lastMs = 0.0
+        avgMs = 0.0
+        peakMs = 0.0
+        lastFpsStampMs = 0L
+        framesSinceFps = 0
+        fpsNow = 0
+    }
+}
+
+private data class ProbeHolder(
+    val track: VideoTrack,
+    val sink: VideoSink,
+    val stats: DecodeProbeSink
+)
+
 /// 自编码解码器（控制端原生实现）。
 ///
 /// 对应 WebRTCController 的 SelfCodecDecoder.java：从 `video` DataChannel 接收
@@ -63,6 +143,10 @@ class SelfCodecDecoderPlugin :
    private var methodChannel: MethodChannel? = null
    private var textureRegistry: TextureRegistry? = null

+    /// WebRTC 全托管模式的解码耗时探针（独立通道，复用同一 MethodCallHandler）。
+    private var probeChannel: MethodChannel? = null
+    private var probeHolder: ProbeHolder? = null
+
    private var textureEntry: TextureRegistry.SurfaceTextureEntry? = null
    private var surfaceTexture: SurfaceTexture? = null
    private var surface: Surface? = null
@@ -77,17 +161,34 @@ class SelfCodecDecoderPlugin :
    private val frameQueue = ArrayList<Frame>()
    private val inputQueue = ArrayList<Int>()

+    // 已完成重组的最大单元序号，用于丢弃明显过期的旧单元。
+    private var lastSeq = -1
+
+    // —— 解码耗时统计 ——
+    // 记录每帧从 queueInputBuffer 到 onOutputBufferAvailable 的时延（毫秒）。
+    private val statsLock = Any()
+    private val decodeStartNs = HashMap<Long, Long>() // pts(us) -> 入队 nanoTime
+    private val recentDecodeMs = ArrayDeque<Double>()  // 最近若干帧耗时，用于滑动平均
+    private var lastDecodeMs = 0.0
+    private var maxDecodeMs = 0.0
+    private var decodedFrames = 0L
+
    private val mainHandler = Handler(Looper.getMainLooper())

    override fun onAttachedToEngine(binding: FlutterPlugin.FlutterPluginBinding) {
        textureRegistry = binding.textureRegistry
        methodChannel = MethodChannel(binding.binaryMessenger, CHANNEL)
        methodChannel?.setMethodCallHandler(this)
+        // 解码耗时探针（WebRTC 全托管模式）：独立通道，复用同一 handler。
+        probeChannel = MethodChannel(binding.binaryMessenger, "webrtc_decode_probe")
+        probeChannel?.setMethodCallHandler(this)
    }

    override fun onDetachedFromEngine(binding: FlutterPlugin.FlutterPluginBinding) {
        methodChannel?.setMethodCallHandler(null)
        methodChannel = null
+        probeChannel?.setMethodCallHandler(null)
+        probeChannel = null
        textureRegistry = null
    }

@@ -126,14 +227,75 @@ class SelfCodecDecoderPlugin :
                if (!enabled) releaseDecoder()
                result.success(null)
            }
+            "getStats" -> {
+                result.success(collectStats())
+            }
            "dispose" -> {
                releaseAll()
                result.success(null)
            }
+            "attachWebRtcDecodeProbe" -> {
+                val trackId = call.argument<String>("trackId")
+                if (trackId == null) {
+                    result.error("PROBE", "trackId is null", null)
+                } else {
+                    attachWebRtcDecodeProbe(trackId, result)
+                }
+            }
+            "detachWebRtcDecodeProbe" -> detachWebRtcDecodeProbe(result)
+            "getWebRtcDecodeStats" -> getWebRtcDecodeStats(result)
            else -> result.notImplemented()
        }
    }

+    // —— WebRTC 全托管模式解码耗时探针 ——
+
+    private fun attachWebRtcDecodeProbe(trackId: String, result: MethodChannel.Result) {
+        try {
+            val plugin = FlutterWebRTCPlugin.sharedSingleton
+            val track = plugin?.getRemoteTrack(trackId)
+            if (track !is VideoTrack) {
+                result.error("PROBE", "remote video track not found: $trackId", null)
+                return
+            }
+            detachWebRtcDecodeProbeInternal()
+            val stats = DecodeProbeSink()
+            val sink = VideoSink { stats.onFrame() }
+            track.addSink(sink)
+            probeHolder = ProbeHolder(track, sink, stats)
+            result.success(null)
+        } catch (e: Exception) {
+            Log.e(TAG, "attachWebRtcDecodeProbe failed", e)
+            result.error("PROBE", e.message, null)
+        }
+    }
+
+    private fun detachWebRtcDecodeProbeInternal() {
+        probeHolder?.let { (track, sink, stats) ->
+            try {
+                track.removeSink(sink)
+            } catch (e: Exception) {
+                Log.w(TAG, "removeSink failed", e)
+            }
+            stats.reset()
+        }
+        probeHolder = null
+    }
+
+    private fun detachWebRtcDecodeProbe(result: MethodChannel.Result) {
+        detachWebRtcDecodeProbeInternal()
+        result.success(null)
+    }
+
+    private fun getWebRtcDecodeStats(result: MethodChannel.Result) {
+        val holder = probeHolder
+        if (holder == null) {
+            result.success(null)
+        } else {
+            result.success(holder.stats.snapshot())
+        }
+    }
+
    private fun createInternal(): Int {
        if (textureEntry != null) return textureEntry!!.id().toInt()
        val entry = textureRegistry?.createSurfaceTexture() ?: return -1
@@ -178,14 +340,26 @@ class SelfCodecDecoderPlugin :
    }

    private fun handleChunk(seq: Int, total: Int, idx: Int, data: ByteArray) {
+        if (total <= 0) return
+        // 过期检查：同一"单元"的所有分片共用同一个 seq（以 idx 区分），
+        // 因此不能用 seq <= lastSeq 丢弃——否则同一单元除首片外的分片都会被误删，
+        // 导致大帧永远无法重组、自编码串流黑屏。这里只丢弃明显更旧（seq 明显小于 lastSeq）
+        // 的旧单元分片，相同 seq（同单元的其它分片）放行。
+        if (lastSeq != -1 && seq < lastSeq && (lastSeq - seq) < 1000) return
        var buf = chunkBuffers[seq]
        if (buf == null) {
            buf = ChunkBuffer(seq, total)
            chunkBuffers[seq] = buf
+            // 防止内存堆积：重组缓冲超过 24 个未完成单元时丢弃最旧的。
+            if (chunkBuffers.size > 24) {
+                val minKey = chunkBuffers.keys.minOrNull()
+                if (minKey != null) chunkBuffers.remove(minKey)
+            }
        }
        buf.add(idx, data)
        if (buf.isComplete()) {
            chunkBuffers.remove(seq)
+            if (seq > lastSeq) lastSeq = seq
            handleUnit(buf.join())
        }
    }
@@ -208,9 +382,21 @@ class SelfCodecDecoderPlugin :

    private fun handleConfig(config: ByteArray) {
        synchronized(configLock) {
-            if (configured) return // 忽略重复参数集
+            // 去重：仅当队列中尚无相同参数集时加入，避免关键帧前重发造成堆积。
+            // 关键：绝不能因为 configured 就 return —— 编码端把 SPS(csd-0) 与 PPS(csd-1)
+            // 作为两个独立的 CONFIG 单元分别发送。若第一个 CONFIG(SPS) 到达后即置
+            // configured=true 而直接丢弃第二个 CONFIG(PPS)，解码器只拿到 SPS 缺少 PPS，
+            // H.264 将无法解码 → 永久黑屏（这正是"自编码无法显示"的根因）。
+            for (c in pendingConfigs) {
+                if (c.contentEquals(config)) return
+            }
            pendingConfigs.add(config)
-            tryConfigure()
+            if (!configured) {
+                tryConfigure()
+            } else {
+                // 已配置时也尝试喂入，以应对中途的参数集更新。
+                feed()
+            }
        }
    }

@@ -240,6 +426,8 @@ class SelfCodecDecoderPlugin :

                    override fun onOutputBufferAvailable(
                        codec: MediaCodec, index: Int, info: MediaCodec.BufferInfo) {
+                        // 用输出帧的 presentationTimeUs 匹配入队时间戳，统计解码耗时。
+                        recordDecodeTime(info.presentationTimeUs)
                        try {
                            codec.releaseOutputBuffer(index, true)
                        } catch (e: Exception) {
@@ -311,6 +499,12 @@ class SelfCodecDecoderPlugin :
                    val flags = if (frame.key) MediaCodec.BUFFER_FLAG_KEY_FRAME else 0
                    decoder?.queueInputBuffer(
                        index, 0, frame.data.size, frame.pts, flags)
+                    // 记录该帧入队时间，供输出回调计算解码耗时。
+                    synchronized(statsLock) {
+                        decodeStartNs[frame.pts] = System.nanoTime()
+                        // 防止乱序/丢帧导致 map 泄漏。
+                        if (decodeStartNs.size > 300) decodeStartNs.clear()
+                    }
                } catch (e: Exception) {
                    Log.w(TAG, "queue frame failed", e)
                }
@@ -318,6 +512,33 @@ class SelfCodecDecoderPlugin :
        }
    }

+    /// 输出帧到达时计算解码耗时（毫秒），并更新滑动平均/峰值/计数。
+    private fun recordDecodeTime(pts: Long) {
+        synchronized(statsLock) {
+            val startNs = decodeStartNs.remove(pts) ?: return
+            val ms = (System.nanoTime() - startNs) / 1_000_000.0
+            lastDecodeMs = ms
+            if (ms > maxDecodeMs) maxDecodeMs = ms
+            decodedFrames++
+            recentDecodeMs.addLast(ms)
+            while (recentDecodeMs.size > DECODE_STATS_WINDOW) recentDecodeMs.removeFirst()
+        }
+    }
+
+    /// 汇总解码统计供 Dart 侧读取（毫秒）。
+    private fun collectStats(): Map<String, Any> {
+        synchronized(statsLock) {
+            val avg = if (recentDecodeMs.isEmpty()) 0.0
+                else recentDecodeMs.sum() / recentDecodeMs.size
+            return mapOf(
+                "lastMs" to lastDecodeMs,
+                "avgMs" to avg,
+                "maxMs" to maxDecodeMs,
+                "count" to decodedFrames
+            )
+        }
+    }
+
    private fun releaseDecoder() {
        synchronized(configLock) {
            try {
@@ -333,6 +554,17 @@ class SelfCodecDecoderPlugin :
            pendingConfigs.clear()
            frameQueue.clear()
            inputQueue.clear()
+            // 重置序号并清空重组缓冲，避免重连/重启后首帧被误判为过期而丢弃。
+            lastSeq = -1
+            chunkBuffers.clear()
+            // 重置解码耗时统计。
+            synchronized(statsLock) {
+                decodeStartNs.clear()
+                recentDecodeMs.clear()
+                lastDecodeMs = 0.0
+                maxDecodeMs = 0.0
+                decodedFrames = 0L
+            }
        }
    }