Measure call video throughput and audio gating losses

src/call/call_video_transport.c

@@ -31,6 +31,9 @@ struct call_video_transport {
     int rate_pending, admit;
     uint64_t received_generation, decode_generation;
     int decode_result, decode_pending, tx_rejected;
+    uint64_t report_us;
+    unsigned offered, accepted, held_busy, transport_busy, key_wait;
+    int64_t max_audio_ahead;
 };
 
 static struct call_session *active(struct UTUN_INSTANCE *inst, uint64_t id) {
@@ -128,6 +131,15 @@ static void update_rate(struct call_video_transport *v) {
     }
     struct route_tx_window produced = {0}; route_tx_meter_window(&v->production, now, &produced);
     int held = v->data && v->tx.pts_us > v->audio_end_pts;
+    if (v->audio_end_pts >= 0 && held && v->tx.pts_us - v->audio_end_pts > v->max_audio_ahead)
+        v->max_audio_ahead = v->tx.pts_us - v->audio_end_pts;
+    if (now - v->report_us >= 1000000) {
+        DEBUG_DEBUG(DEBUG_CATEGORY_CALL,
+            "video TX flow call=%llu offered=%u accepted=%u audio_busy=%u transport_busy=%u key_wait=%u audio_ahead_us=%lld held=%d",
+            (unsigned long long)s->call_id, v->offered, v->accepted, v->held_busy, v->transport_busy, v->key_wait,
+            (long long)v->max_audio_ahead, held);
+        v->report_us = now; v->offered = v->accepted = v->held_busy = v->transport_busy = v->key_wait = 0; v->max_audio_ahead = 0;
+    }
     struct call_video_rate_sample sample = {0};
     sample.now_us = now; sample.window_end_us = now / ROUTE_TX_SAMPLE_US * ROUTE_TX_SAMPLE_US;
     sample.delivered_bytes = media_bytes(net.window.bytes, net.window.packets); sample.produced_bytes = produced.bytes;
@@ -263,7 +275,11 @@ int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_fr
         }
         if (!v->admit) { v->need_key = 1; return 0; }
     }
-    if (v->data) { v->need_key = 1; return 0; }
+    v->offered++;
+    if (v->data) {
+        if (v->tx.pts_us > v->audio_end_pts) v->held_busy++; else v->transport_busy++;
+        v->need_key = 1; return 0;
+    }
     struct ETCP_ROUTER_TX_STATS net;
     if (etcp_router_tx_stats(inst, s->group_id, s->peer_node_id, ETCP_RT_ID_CALL_VIDEO, get_time_us(), &net) == 0) {
         uint64_t debt = media_bytes(net.queued_bytes + net.inflight_bytes, (uint64_t)net.queued_packets + net.inflight_packets);
@@ -274,12 +290,12 @@ int call_video_send_frame(struct UTUN_INSTANCE *inst, const struct call_video_fr
         DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video TX format changed without new generation"); return -1;
     }
     if (changed || (v->tx_last_id && f->frame_id != v->tx_last_id + 1)) v->need_key = 1;
-    if (v->need_key && !(f->flags & CALL_VIDEO_KEY)) return 0;
+    if (v->need_key && !(f->flags & CALL_VIDEO_KEY)) { v->key_wait++; return 0; }
     v->data = u_malloc(f->size);
     if (!v->data) { DEBUG_ERROR(DEBUG_CATEGORY_CALL, "video TX allocation failed bytes=%u", f->size); return -1; }
     memcpy(v->data, f->data, f->size); v->tx = *f; v->tx.data = v->data;
     v->tx_started_tb = get_time_tb(); v->tx_last_id = f->frame_id; v->tx_last_pts = f->pts_us; v->need_key = 0;
-    flush(v); return 1;
+    v->accepted++; flush(v); return 1;
 }
 int call_video_configure_rate(struct UTUN_INSTANCE *inst, uint64_t id, const struct call_video_format *format,
                               uint32_t applied, uint32_t maximum) {

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

@@ -31,6 +31,8 @@ internal class CallVideoPlayer(private val id: Long, private val failure: (Strin
         var held = -1; var heldPts = -1L; var inflight = 0; var draining = false
         var sourceEpoch = 0L; 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()
         fun close() {
             try { decoder?.stop() } catch (e: Exception) { LogManager.addLog("WARN", "call", "video decoder stop: ${e.message}") }
@@ -48,6 +50,10 @@ internal class CallVideoPlayer(private val id: Long, private val failure: (Strin
                         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}")
@@ -57,9 +63,10 @@ internal class CallVideoPlayer(private val id: Long, private val failure: (Strin
                         val render = heldPts >= clock[4] - 500_000
                         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) {
@@ -75,10 +82,13 @@ internal class CallVideoPlayer(private val id: Long, private val failure: (Strin
                                 } }
                                 lastOutput = info.presentationTimeUs; inflight = (inflight - 1).coerceAtLeast(0)
                                 held = out; heldPts = info.presentationTimeUs; lastProgress = now
+                                decoded++
                             }
                         }
                     }
-                    if (pending == null && held < 0 && inflight < 8) pending = CallVideoNative.pull(id, meta)
+                    if (pending == null && held < 0 && inflight < 8) {
+                        pending = CallVideoNative.pull(id, meta); if (pending != null) pulled++
+                    }
                     val data = pending
                     if (data != null && surface?.isValid == true && held < 0) {
                         val next = CallVideoFormat.from(meta)
@@ -109,7 +119,7 @@ internal class CallVideoPlayer(private val id: Long, private val failure: (Strin
                                 } else data
                                 val buffer = checkNotNull(decoder!!.getInputBuffer(index)); check(bytes.size <= buffer.capacity())
                                 buffer.put(bytes); decoder!!.queueInputBuffer(index, 0, bytes.size, meta[10], 0)
-                                pending = null; inflight++
+                                pending = null; inflight++; queued++
                             }
                         }
                     } else if (data != null && surface == null) { pending = null; CallVideoNative.requestKey(id) }
@@ -118,6 +128,14 @@ internal class CallVideoPlayer(private val id: Long, private val failure: (Strin
                         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")
+                        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