diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallVideoNative.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallVideoNative.kt index 731bca6d..80cf5763 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallVideoNative.kt +++ b/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? diff --git a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallVideoPlayer.kt b/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallVideoPlayer.kt index b506da02..d04a1a75 100644 --- a/tools/chatgui-android/app/src/main/java/com/utun/chat/data/CallVideoPlayer.kt +++ b/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 held = -1; var heldPts = -1L; var inflight = 0; var draining = false - var sourceEpoch = 0L; var lastOutput = -1L; var lastProgress = System.nanoTime() + var pending: ByteArray? = null; val meta = LongArray(13) + val pictures = ArrayDeque(); val inputs = ArrayDeque() + 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 - if (!valid) invalidClock++ - 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) { + // Два готовых output нужны для пропуска старого показа после паузы CPU, не для дополнительного запаса. + while (decoder != null && pictures.size < 2 && !endSeen) { val out = decoder!!.dequeueOutputBuffer(info, 0) if (out >= 0) { - if (info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0) { - decoder!!.releaseOutputBuffer(out, false); close() - } else if (info.size == 0) decoder!!.releaseOutputBuffer(out, false) + worked = true; lastProgress = System.nanoTime() + if (info.flags and MediaCodec.BUFFER_FLAG_END_OF_STREAM != 0) endSeen = true + 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) - held = out; heldPts = info.presentationTimeUs; lastProgress = now - if (heldPts % 1_000_000 < 34_000) LogManager.addLog("DEBUG", "debug", - "video timing stage=decode call=$id pts=$heldPts mono_us=${System.nanoTime() / 1000} " + - "audio_pts=${clock[4]} audio_valid=$valid inflight=$inflight") + val timing = checkNotNull(inputs.pollFirst()) { "Video output without input timing" } + check(timing.pts == info.presentationTimeUs) { "Video output timing mismatch" } + lastOutput = info.presentationTimeUs; pictures.addLast(Picture(out, timing)) + if (timing.pts % 1_000_000 < 34_000) LogManager.addLog("DEBUG", "debug", + "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 " + - "pending=${pending != null} held=$held aheadUs=$maxAheadUs lateUs=$maxLateUs " + - "invalidClock=$invalidClock audioPts=${clock[4]} videoPts=$heldPts") + "pulled=$pulled queued=$queued decoded=$decoded rendered=$rendered skipped=$late inflight=${inputs.size} " + + "pending=${pending != null} ready=${pictures.size} audioValid=$valid audioPts=${clock[4]}") 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() } } } diff --git a/tools/chatgui-android/jni_bridge/android_call_video_jni.c b/tools/chatgui-android/jni_bridge/android_call_video_jni.c index ee6d4fd8..18261940 100644 --- a/tools/chatgui-android/jni_bridge/android_call_video_jni.c +++ b/tools/chatgui-android/jni_bridge/android_call_video_jni.c @@ -2,12 +2,14 @@ #include #include #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); }