Measure stream viewer request and worker delays

src/stream/stream_viewer.c

@@ -19,6 +19,8 @@ struct viewer {
     int64_t position, audio_ahead;
     uint64_t replies, fulfilled_repeats; /* Выданные ответы и погашенные ими повторные запросы. */
     uint64_t created_us, prepared_us, report_us;
+    /* Время подготовки ответа и его ожидания до pull; статистика за отчётный интервал. */
+    uint64_t demand_us, delivered, prepare_total_us, prepare_max_us, handoff_total_us, handoff_max_us;
     unsigned first_records; /* Диагностика первой выдачи VIDEO/AUDIO; не влияет на cursor. */
 };
 struct stream_viewers {
@@ -129,17 +131,20 @@ static void process(void *arg) {
         uint64_t now = get_time_us();
         if (now - s->report_us >= 1000000) {
             s->report_us = now;
-            DEBUG_DEBUG(DEBUG_CATEGORY_STREAM,
+            DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG,
                 "viewer buffer id=%llu source=%llx epoch=%llx elapsed_us=%llu initialized=%d generation=%llu "
                 "start_us=%lld position_us=%lld ready_us=%lld video_us=%lld audio_us=%lld reserve_us=%lld "
-                "requested=%d waiting=%d slot=%d slot_age_us=%llu",
+                "requested=%d waiting=%d slot=%d slot_age_us=%llu records=%llu prepare_us=%llu/%llu handoff_us=%llu/%llu timing=mean/max",
                 (unsigned long long)s->id, (unsigned long long)s->source, (unsigned long long)s->epoch,
                 (unsigned long long)(now - s->created_us), s->status.initialized, (unsigned long long)s->status.generation,
                 (long long)s->status.start_us, (long long)s->position, (long long)s->status.ready_us,
                 (long long)s->status.video_us, (long long)s->status.audio_us,
                 (long long)(s->status.initialized && s->position >= 0 ? s->status.ready_us - s->position : -1),
                 s->requested, s->waiting, s->record ? s->record->kind : 0,
-                (unsigned long long)(s->record ? now - s->prepared_us : 0));
+                (unsigned long long)(s->record ? now - s->prepared_us : 0), (unsigned long long)s->delivered,
+                (unsigned long long)(s->delivered ? s->prepare_total_us / s->delivered : 0), (unsigned long long)s->prepare_max_us,
+                (unsigned long long)(s->delivered ? s->handoff_total_us / s->delivered : 0), (unsigned long long)s->handoff_max_us);
+            s->delivered = s->prepare_total_us = s->prepare_max_us = s->handoff_total_us = s->handoff_max_us = 0;
         }
         if (!s->requested || s->waiting || s->record) continue;
         s->requested = 0;
@@ -231,6 +236,13 @@ struct stream_viewer_record *stream_viewer_pull(struct stream_viewers *v, uint64
     struct stream_viewer_record *r = s && !s->stopping ? s->record : NULL;
     if (r) {
         unsigned bit = r->kind == STREAM_VIEWER_VIDEO ? 1 : r->kind == STREAM_VIEWER_AUDIO ? 2 : 0;
+        if (bit) {
+            uint64_t handoff = get_time_us() - s->prepared_us;
+            uint64_t prepare = s->demand_us && s->prepared_us >= s->demand_us ? s->prepared_us - s->demand_us : 0;
+            s->delivered++; s->prepare_total_us += prepare; s->handoff_total_us += handoff;
+            if (prepare > s->prepare_max_us) s->prepare_max_us = prepare;
+            if (handoff > s->handoff_max_us) s->handoff_max_us = handoff;
+        }
         if (bit && !(s->first_records & bit)) {
             uint64_t now = get_time_us(); s->first_records |= bit;
             DEBUG_INFO(DEBUG_CATEGORY_STREAM,
@@ -263,6 +275,7 @@ int stream_viewer_request(struct stream_viewers *v, uint64_t id, int64_t positio
     int result = -1;
     if (s && !s->stopping && (s->opening || s->handle)) {
         int changed = s->position != position || s->video != video || s->audio_ahead != audio_ahead;
+        if (!s->requested && !s->record) s->demand_us = get_time_us();
         s->position = position; s->video = video; s->audio_ahead = audio_ahead; s->requested = video || audio_ahead >= 0;
         result = 0;
         if (s->requested && !s->record && (!s->waiting || changed)) { s->waiting = 0; result = schedule(v); }
@@ -287,6 +300,7 @@ int stream_viewer_resync(struct stream_viewers *v, uint64_t id) {
     if (s && s->handle && !s->stopping && schedule(v) == 0) {
         u_free(s->record); s->record = NULL;
         memset(&s->status, 0, sizeof(s->status));
+        s->demand_us = get_time_us();
         s->resync = s->requested = 1; s->waiting = 0; s->position = -1; result = 0;
     }
     pthread_mutex_unlock(&v->lock); return result;

tools/chatgui/src/streamplayer_window.cpp

@@ -293,7 +293,7 @@ void StreamPlayerWindow::decode() {
     /* Забираем выход прежде следующего входа. При FIN тот же проход дренирует codec до EOF. */
     auto drainVideo = [&] {
         if (!decoder) return;
-        for (unsigned work = 0; work < 8 && p.running; work++) {
+        for (unsigned attempt = 0; attempt < 8 && p.running; attempt++) {
             { std::lock_guard<std::mutex> guard(p.lock); if (p.videos.size() >= 2) return; }
             int64_t codecAt = nowUs();
             int result = avcodec_receive_frame(decoder, frame);

tools/chatgui/tests/test_stream_view.cpp

@@ -208,6 +208,7 @@ private slots:
     void initTestCase() {
         debug_config_init(); debug_set_level(DEBUG_LEVEL_WARN); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
         if (qEnvironmentVariableIsSet("UTUN_TEST_DEBUG")) {
+            debug_set_category_level(DEBUG_CATEGORY_DEBUG, DEBUG_LEVEL_DEBUG);
             debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_TRACE);
             debug_set_category_level(DEBUG_CATEGORY_VIDEO, DEBUG_LEVEL_TRACE);
         }