Use shared video playout on Android without blocking decoder input while waiting

tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallVideoNative.kt

@@ -12,6 +12,12 @@ internal object CallVideoNative {
     external fun applied(id: Long, generation: Long, request: Long, bps: Int)
     external fun send(id: Long, meta: LongArray, bytes: ByteArray): Int
     external fun pull(id: Long, meta: LongArray): ByteArray?
+    external fun playoutCreate(id: Long): Long
+    external fun playoutDestroy(handle: Long)
+    external fun playoutReset(handle: Long)
+    /** >0: ждать микросекунды; 0: показать; <0: пропустить только показ. */
+    external fun playoutStep(handle: Long, pts: Long, received: Long, nextPts: Long, nextReceived: Long,
+        now: Long, valid: Boolean, audio: Long): Long
     external fun requestKey(id: Long)
     external fun range(id: Long, minimum: Int, maximum: Int): Boolean
     external fun inspect(codec: Int, bytes: ByteArray): LongArray?

tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallVideoPlayer.kt

@@ -8,8 +8,9 @@ import android.view.Surface
 import kotlinx.coroutines.flow.MutableStateFlow
 import kotlinx.coroutines.flow.StateFlow
 import java.util.concurrent.ConcurrentLinkedQueue
+import java.util.ArrayDeque
 
-/** Один RX worker; удерживаем output buffer до фактического audio PTS. Новое поколение ждёт drain старого. */
+/** Один RX worker, общий абсолютный таймер показа. Новое поколение ждёт drain старого. */
 internal class CallVideoPlayer(private val id: Long, private val failure: (String) -> Unit) {
     @Volatile private var running = true
     private data class Output(val surface: Surface?, val release: () -> Unit, val detach: Surface? = null)
@@ -28,83 +29,90 @@ internal class CallVideoPlayer(private val id: Long, private val failure: (Strin
     fun stop() { synchronized(this) { running = false }; worker.join() }
 
     private fun loop() {
+        data class Timing(val pts: Long, val received: Long)
+        data class Picture(val index: Int, val timing: Timing)
+        val playout = CallVideoNative.playoutCreate(id)
+        if (playout == 0L) { failure("Video playout allocation failed"); return }
         var decoder: MediaCodec? = null; var surface: Surface? = null; var format: CallVideoFormat? = null
         var output: Output? = null; var lastRendered = 0L
-        var pending: ByteArray? = null; var meta = LongArray(12)
+        var pending: ByteArray? = null; val meta = LongArray(13)
-        var held = -1; var heldPts = -1L; var inflight = 0; var draining = false
+        val pictures = ArrayDeque<Picture>(); val inputs = ArrayDeque<Timing>()
-        var sourceEpoch = 0L; var lastOutput = -1L; var lastProgress = System.nanoTime()
+        var draining = false; var endSeen = false
+        var lastOutput = -1L; var lastProgress = System.nanoTime()
         var reportedGeneration = 0L; var rejectedGeneration = 0L
         var reportAt = System.nanoTime(); var pulled = 0; var queued = 0; var decoded = 0; var rendered = 0; var late = 0
-        var maxAheadUs = 0L; var maxLateUs = 0L; var invalidClock = 0
         val clock = LongArray(8); val info = MediaCodec.BufferInfo()
         val renderHandler = Handler(Looper.getMainLooper())
         fun close() {
             try { decoder?.stop() } catch (e: Exception) { LogManager.addLog("WARN", "call", "video decoder stop: ${e.message}") }
-            decoder?.release(); decoder = null; format = null; held = -1; inflight = 0; draining = false; lastOutput = -1
+            decoder?.release(); decoder = null; format = null; pictures.clear(); inputs.clear()
+            draining = false; endSeen = false; lastOutput = -1
         }
         try {
             while (running) {
+                var worked = false; var waitUs = 5000L
                 try {
-                    val now = System.nanoTime()
                     while (true) {
                         val next = outputs.poll() ?: break
                         if (next.detach != null && output?.surface !== next.detach) continue
                         close(); output?.release?.invoke(); output = next; surface = next.surface
+                        CallVideoNative.playoutReset(playout)
                         pending = null; displayed.value = null; CallVideoNative.requestKey(id)
                         LogManager.addLog("INFO", "call", "video output attached=${surface != null} call=$id")
                     }
-                    val valid = NativeLib.callAudioClock(id, now / 1000, clock) && clock[5] != 0L
+                    // Два готовых output нужны для пропуска старого показа после паузы CPU, не для дополнительного запаса.
-                    if (!valid) invalidClock++
+                    while (decoder != null && pictures.size < 2 && !endSeen) {
-                    if (held >= 0 && valid) {
-                        maxAheadUs = maxOf(maxAheadUs, heldPts - clock[4]); maxLateUs = maxOf(maxLateUs, clock[4] - heldPts)
-                    }
-                    if (valid && sourceEpoch != clock[1]) {
-                        if (sourceEpoch != 0L) LogManager.addLog("DEBUG", "call",
-                            "video audio clock epoch $sourceEpoch -> ${clock[1]}; decoder retained generation=${format?.generation}")
-                        sourceEpoch = clock[1]
-                    }
-                    if (held >= 0 && valid && heldPts <= clock[4]) {
-                        val render = heldPts >= clock[4] - 500_000
-                        if (heldPts % 1_000_000 < 34_000) LogManager.addLog("DEBUG", "debug",
-                            "video timing stage=release call=$id pts=$heldPts mono_us=${System.nanoTime() / 1000} " +
-                                "audio_pts=${clock[4]} render=$render")
-                        decoder!!.releaseOutputBuffer(held, render); held = -1
-                        if (render) {
-                            rendered++
-                            if (displayed.value != format) LogManager.addLog("INFO", "call", "video visible call=$id format=$format")
-                            displayed.value = format; lastRendered = now
-                        } else late++
-                        lastProgress = now
-                    }
-                    if (decoder != null && held < 0) {
                         val out = decoder!!.dequeueOutputBuffer(info, 0)
                         if (out >= 0) {
-                            if (info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0) {
+                            worked = true; lastProgress = System.nanoTime()
-                                decoder!!.releaseOutputBuffer(out, false); close()
+                            if (info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0) endSeen = true
-                            } else if (info.size == 0) decoder!!.releaseOutputBuffer(out, false)
+                            if (info.size == 0) decoder!!.releaseOutputBuffer(out, false)
                             else {
                                 check(info.presentationTimeUs > lastOutput) { "Video decoder reordered PTS" }
                                 format?.let { f -> if (reportedGeneration != f.generation) {
                                     CallVideoNative.decoded(id, f.generation, true); reportedGeneration = f.generation
                                 } }
-                                lastOutput = info.presentationTimeUs; inflight = (inflight - 1).coerceAtLeast(0)
+                                val timing = checkNotNull(inputs.pollFirst()) { "Video output without input timing" }
-                                held = out; heldPts = info.presentationTimeUs; lastProgress = now
+                                check(timing.pts == info.presentationTimeUs) { "Video output timing mismatch" }
-                                if (heldPts % 1_000_000 < 34_000) LogManager.addLog("DEBUG", "debug",
+                                lastOutput = info.presentationTimeUs; pictures.addLast(Picture(out, timing))
-                                    "video timing stage=decode call=$id pts=$heldPts mono_us=${System.nanoTime() / 1000} " +
+                                if (timing.pts % 1_000_000 < 34_000) LogManager.addLog("DEBUG", "debug",
-                                        "audio_pts=${clock[4]} audio_valid=$valid inflight=$inflight")
+                                    "video timing stage=decode call=$id pts=${timing.pts} mono_us=${System.nanoTime() / 1000} " +
+                                        "received_us=${timing.received} inflight=${inputs.size}")
                                 decoded++
                             }
-                        } else if (out == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED)
+                        } else if (out == MediaCodec.INFO_OUTPUT_FORMAT_CHANGED) {
                             LogManager.addLog("INFO", "call", "video decoder output call=$id format=${decoder!!.outputFormat}")
+                        } else break
+                    }
+                    val now = System.nanoTime()
+                    val valid = NativeLib.callAudioClock(id, now / 1000, clock) && clock[5] != 0L
+                    val first = pictures.peekFirst(); val nextPicture = if (pictures.size > 1) pictures.peekLast() else null
+                    val decision = CallVideoNative.playoutStep(playout, first?.timing?.pts ?: -1, first?.timing?.received ?: 0,
+                        nextPicture?.timing?.pts ?: -1, nextPicture?.timing?.received ?: 0, now / 1000, valid, clock[4])
+                    if (decision > 0) waitUs = decision
+                    else if (first != null) {
+                        val render = decision == 0L
+                        if (first.timing.pts % 1_000_000 < 34_000) LogManager.addLog("DEBUG", "debug",
+                            "video timing stage=release call=$id pts=${first.timing.pts} mono_us=${System.nanoTime() / 1000} " +
+                                "audio_pts=${clock[4]} render=$render decision=$decision")
+                        decoder!!.releaseOutputBuffer(first.index, render); pictures.removeFirst(); worked = true
+                        if (render) {
+                            rendered++
+                            if (displayed.value != format) LogManager.addLog("INFO", "call", "video visible call=$id format=$format")
+                            displayed.value = format; lastRendered = now
+                        } else late++
+                        lastProgress = now
                     }
-                    if (pending == null && held < 0 && inflight < 8) {
+                    if (endSeen && pictures.isEmpty()) close()
+                    // Ожидание срока output не останавливает подачу input; очередь остаётся ограниченной.
+                    if (pending == null && inputs.size + pictures.size < 8) {
                         pending = CallVideoNative.pull(id, meta); if (pending != null) pulled++
                         if (pending != null && meta[10] % 1_000_000 < 34_000) LogManager.addLog("DEBUG", "debug",
                             "video timing stage=pull call=$id frame=${meta[9]} pts=${meta[10]} mono_us=${System.nanoTime() / 1000} " +
                                 "boot_us=${SystemClock.elapsedRealtimeNanos() / 1000} audio_pts=${clock[4]} audio_valid=$valid")
                     }
                     val data = pending
-                    if (data != null && surface?.isValid == true && held < 0) {
+                    if (data != null && surface?.isValid == true) {
                         val next = CallVideoFormat.from(meta)
                         if (next.generation == rejectedGeneration) { pending = null; continue }
                         val key = meta[11] != 0L
@@ -128,7 +136,7 @@ internal class CallVideoPlayer(private val id: Long, private val failure: (Strin
                                         "video timing stage=surface call=$id generation=${next.generation} pts=$pts " +
                                             "reported_us=${at / 1000} callback_us=${System.nanoTime() / 1000}")
                                 }, renderHandler)
-                                decoder!!.start(); format = next; lastProgress = now
+                                decoder!!.start(); format = next; lastProgress = System.nanoTime()
                                 LogManager.addLog("INFO", "call", "video decoder started call=$id codec=${implementation.name} format=$next")
                             }
                             val index = decoder!!.dequeueInputBuffer(0)
@@ -141,33 +149,33 @@ internal class CallVideoPlayer(private val id: Long, private val failure: (Strin
                                 buffer.put(bytes); decoder!!.queueInputBuffer(index, 0, bytes.size, meta[10], 0)
                                 if (meta[10] % 1_000_000 < 34_000) LogManager.addLog("DEBUG", "debug",
                                     "video timing stage=queue call=$id frame=${meta[9]} pts=${meta[10]} mono_us=${System.nanoTime() / 1000}")
-                                pending = null; inflight++; queued++
+                                inputs.addLast(Timing(meta[10], meta[12])); pending = null; queued++; worked = true
                             }
                         }
                     } else if (data != null && surface == null) { pending = null; CallVideoNative.requestKey(id) }
-                    if (decoder != null && held < 0 && inflight > 0 && now - lastProgress > 2_000_000_000) error("Video decoder stalled")
+                    if (decoder != null && pictures.isEmpty() && inputs.isNotEmpty() && now - lastProgress > 2_000_000_000)
+                        error("Video decoder stalled")
                     if (lastRendered != 0L && now - lastRendered >= 5_000_000_000) {
                         displayed.value = null; lastRendered = 0
                         LogManager.addLog("INFO", "call", "video hidden: no rendered frames for 5s call=$id")
                     }
                     if (now - reportAt >= 1_000_000_000) {
                         LogManager.addLog("DEBUG", "call", "video playback call=$id intervalMs=${(now - reportAt) / 1_000_000} " +
-                            "pulled=$pulled queued=$queued decoded=$decoded rendered=$rendered late=$late inflight=$inflight " +
+                            "pulled=$pulled queued=$queued decoded=$decoded rendered=$rendered skipped=$late inflight=${inputs.size} " +
-                            "pending=${pending != null} held=$held aheadUs=$maxAheadUs lateUs=$maxLateUs " +
+                            "pending=${pending != null} ready=${pictures.size} audioValid=$valid audioPts=${clock[4]}")
-                            "invalidClock=$invalidClock audioPts=${clock[4]} videoPts=$heldPts")
                         reportAt = now; pulled = 0; queued = 0; decoded = 0; rendered = 0; late = 0
-                        maxAheadUs = 0; maxLateUs = 0; invalidClock = 0
                     }
                 } catch (e: Exception) {
                     LogManager.addLog("ERROR", "call", "video receive call=$id: ${e.message}")
                     val generation = if (pending != null) meta[0] else format?.generation ?: 0
                     if (generation != 0L) { CallVideoNative.decoded(id, generation, false); rejectedGeneration = generation }
-                    close(); pending = null; displayed.value = null; failure(e.message ?: "Video decoder failed")
+                    close(); CallVideoNative.playoutReset(playout)
+                    pending = null; displayed.value = null; failure(e.message ?: "Video decoder failed")
                 }
-                Thread.sleep(5)
+                if (!worked) Thread.sleep(waitUs / 1000, (waitUs % 1000 * 1000).toInt())
             }
         } finally {
-            close(); displayed.value = null; output?.release?.invoke()
+            close(); CallVideoNative.playoutDestroy(playout); displayed.value = null; output?.release?.invoke()
             while (true) { val next = outputs.poll() ?: break; next.release() }
         }
     }

tools/chatgui-android/jni_bridge/android_call_video_jni.c

@@ -2,12 +2,14 @@
 #include <jni.h>
 #include <string.h>
 #include "../../../src/call/call_video_port.h"
+#include "../../../src/call/call_video_playout.h"
 #include "../../../src/call/call_video_bitstream.h"
 #include "../../../lib/debug_config.h"
 #include "../../../lib/mem.h"
 
 /* LongArray format: generation,width,height,fps,rotation,codec,level,maximum_bps,profile.
- * Frame metadata добавляет frame_id,pts_us,flags. JNI не хранит заимствованные массивы. */
+ * TX добавляет frame_id,pts_us,flags; RX также received_us (локальный MONOTONIC).
+ * JNI не хранит заимствованные массивы. */
 static int read_format(JNIEnv *e, jlongArray array, struct call_video_format *f, jlong values[12], int frame) {
     if (!array || (*e)->GetArrayLength(e, array) != (frame ? 12 : 9)) goto invalid;
     (*e)->GetLongArrayRegion(e, array, 0, frame ? 12 : 9, values);
@@ -81,16 +83,42 @@ JNIEXPORT jint JNICALL Java_com_utun_chat_data_CallVideoNative_send(
 }
 JNIEXPORT jbyteArray JNICALL Java_com_utun_chat_data_CallVideoNative_pull(JNIEnv *e, jobject o, jlong id, jlongArray meta) {
     (void)o;
-    if (!meta || (*e)->GetArrayLength(e, meta) != 12) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "call video JNI invalid pull metadata"); return NULL; }
+    if (!meta || (*e)->GetArrayLength(e, meta) != 13) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "call video JNI invalid pull metadata"); return NULL; }
     struct call_video_frame f;
     if (call_video_port_pull(id, &f) != 1) return NULL;
     jlong values[] = {f.format.generation, f.format.width, f.format.height, f.format.fps, f.format.rotation,
-        f.format.codec, f.format.level, f.format.maximum_bps, f.format.profile, f.frame_id, f.pts_us, f.flags};
+        f.format.codec, f.format.level, f.format.maximum_bps, f.format.profile, f.frame_id, f.pts_us, f.flags, f.received_us};
     jbyteArray out = (*e)->NewByteArray(e, f.size);
-    if (out) { (*e)->SetByteArrayRegion(e, out, 0, f.size, (jbyte *)f.data); (*e)->SetLongArrayRegion(e, meta, 0, 12, values); }
+    if (out) { (*e)->SetByteArrayRegion(e, out, 0, f.size, (jbyte *)f.data); (*e)->SetLongArrayRegion(e, meta, 0, 13, values); }
     else call_video_port_request_key(id);
     u_free((void *)f.data); return out;
 }
+/* Handle принадлежит только RX worker, включая reset/destroy. */
+JNIEXPORT jlong JNICALL Java_com_utun_chat_data_CallVideoNative_playoutCreate(JNIEnv *e, jobject o, jlong id) {
+    (void)e; (void)o;
+    struct call_video_playout *p = u_malloc(sizeof(*p));
+    if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video playout allocation failed"); return 0; }
+    call_video_playout_init(p, id); return (jlong)(uintptr_t)p;
+}
+JNIEXPORT void JNICALL Java_com_utun_chat_data_CallVideoNative_playoutDestroy(JNIEnv *e, jobject o, jlong handle) {
+    (void)e; (void)o; u_free((void *)(uintptr_t)handle);
+}
+JNIEXPORT void JNICALL Java_com_utun_chat_data_CallVideoNative_playoutReset(JNIEnv *e, jobject o, jlong handle) {
+    (void)e; (void)o; struct call_video_playout *p = (void *)(uintptr_t)handle;
+    if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video playout reset without handle"); return; }
+    call_video_playout_init(p, p->call_id);
+}
+JNIEXPORT jlong JNICALL Java_com_utun_chat_data_CallVideoNative_playoutStep(JNIEnv *e, jobject o, jlong handle,
+    jlong pts, jlong received, jlong next_pts, jlong next_received, jlong now, jboolean valid, jlong audio) {
+    (void)e; (void)o; struct call_video_playout *p = (void *)(uintptr_t)handle;
+    if (!p) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video playout poll without handle"); return 5000; }
+    struct call_video_playout_frame f = {pts, received}, next = {next_pts, next_received};
+    int action = call_video_playout_poll(p, pts >= 0 ? &f : NULL, next_pts >= 0 ? &next : NULL, now, valid, audio);
+    if (action == CALL_VIDEO_PLAYOUT_SHOW) return 0;
+    if (action != CALL_VIDEO_PLAYOUT_WAIT) return -action;
+    int64_t wait = p->deadline_us - now;
+    return wait < 100 ? 100 : wait > 5000 ? 5000 : wait;
+}
 JNIEXPORT void JNICALL Java_com_utun_chat_data_CallVideoNative_requestKey(JNIEnv *e, jobject o, jlong id) {
     (void)e; (void)o; call_video_port_request_key(id);
 }