Consume repeated viewer demand when delivering its reply

src/stream/stream_viewer.c

@@ -17,6 +17,7 @@ struct viewer {
     int opening, stopping, resync, requested, waiting;
     int video;
     int64_t position, audio_ahead;
+    uint64_t replies, fulfilled_repeats; /* Выданные ответы и погашенные ими повторные запросы. */
 };
 struct stream_viewers {
     pthread_mutex_t lock;
@@ -66,8 +67,9 @@ static void close_viewer(struct viewer *s, enum stream_viewer_kind kind, int rea
     if (s->handle) stream_close(s->handle);
     s->handle = NULL; s->opening = s->requested = s->waiting = s->resync = 0;
     if (kind) output(s, kind, reason, NULL);
-    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer released id=%llu group=%llx source=%llx kind=%d reason=%d",
-        (unsigned long long)s->id, (unsigned long long)s->group, (unsigned long long)s->source, kind, reason);
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "viewer released id=%llu group=%llx source=%llx kind=%d reason=%d replies=%llu repeats=%llu",
+        (unsigned long long)s->id, (unsigned long long)s->group, (unsigned long long)s->source, kind, reason,
+        (unsigned long long)s->replies, (unsigned long long)s->fulfilled_repeats);
 }
 
 /* Событие будит всех readers этой эпохи, не меняя их независимые позиции и запросы. */
@@ -207,7 +209,16 @@ struct stream_viewer_record *stream_viewer_pull(struct stream_viewers *v, uint64
     pthread_mutex_lock(&v->lock);
     struct viewer *s = find(v, id);
     struct stream_viewer_record *r = s && !s->stopping ? s->record : NULL;
-    if (r) s->record = NULL;
+    if (r) {
+        /* Ответ обслуживает и повторный request: AVAILABLE не должен прочитать следующий кадр до нового demand. */
+        int repeated = s->requested;
+        s->record = NULL; s->requested = 0; s->replies++;
+        if (repeated) s->fulfilled_repeats++;
+        if (s->replies == 1 || (repeated && (s->fulfilled_repeats == 1 || s->fulfilled_repeats % 256 == 0)))
+            DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "viewer reply id=%llu kind=%d pts=%llu repeated=%d replies=%llu repeats=%llu pending=0",
+                (unsigned long long)s->id, r->kind, (unsigned long long)r->pts, repeated,
+                (unsigned long long)s->replies, (unsigned long long)s->fulfilled_repeats);
+    }
     pthread_mutex_unlock(&v->lock); return r;
 }
 

src/stream/stream_viewer.h

@@ -30,7 +30,8 @@ void stream_viewers_destroy(struct stream_viewers *viewers);
  * Открытие асинхронно: OPENED/CLOSED возвращается через pull. tracks — VIDEO или VIDEO|AUDIO. */
 int stream_viewer_open(struct stream_viewers *viewers, uint64_t id, uint64_t group, uint64_t source, uint64_t epoch, unsigned tracks);
 void stream_viewer_stop(struct stream_viewers *viewers, uint64_t id);
-/* Один owned ответ или NULL. Новые данные запрашиваются request; idle не копирует кольцо. */
+/* Один owned ответ или NULL. Выдача ответа погашает demand, включая повторный request до pull.
+ * NULL сохраняет ожидающий запрос. Следующие данные требуют нового request; idle не копирует кольцо. */
 struct stream_viewer_record *stream_viewer_pull(struct stream_viewers *viewers, uint64_t id);
 /* Позиция и audio_ahead относятся только к этому reader; повторный demand заменяет прежний.
  * 0 — принято, -1 — неизвестный/закрытый ID или ошибка scheduling. */

tests/test_android_stream.c

@@ -173,20 +173,48 @@ int main(void) {
     CHECK(stream_member_load(inst, group, inst->node_id, &after_switch, NULL) == STREAM_OK);
     CHECK(after_switch.published && after_switch.source.source_epoch == member.source.source_epoch);
     /* Повторные запросы готовят один ответ; получение ответа не создаёт скрытую следующую копию. */
+    debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_DEBUG);
     for (unsigned i = 0; i < 5; i++)
         CHECK(android_stream_feed(1, 1, 800000 + i * 40000, 0, video, sizeof(video), &next_format) == 1);
     for (unsigned i = 0; i < 50; i++) { CHECK(android_stream_read_request(2, 700000, 1, -1) == 0); test_clock_poll(ua, 100); }
     output = android_stream_pull(2, &size); CHECK(output && output[0] == 1 && stream_read_be(output + 8, 8) == 800000); u_free(output);
+    /* AVAILABLE между pull и отменой demand раньше выдавал второй кадр при занятом decoder input. */
+    CHECK(android_stream_feed(1, 1, 1000000, 0, video, sizeof(video), &next_format) == 1);
+    test_clock_run_for(ua, 1000);
+    output = android_stream_pull(2, &size);
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "demand regression: new data after reply, unsolicited_kind=%d", output ? output[0] : 0);
+    CHECK(!output);
     test_clock_run_for(ua, 1000); CHECK(!android_stream_pull(2, &size));
     CHECK(android_stream_read_request(2, 700000, 0, -1) == 0); test_clock_run_for(ua, 1000);
     CHECK(!android_stream_pull(2, &size));
-    for (unsigned frame = 1; frame < 5; frame++) {
+    for (unsigned frame = 1; frame < 6; frame++) {
         output = NULL;
         for (unsigned i = 0; i < 1000 && !output; i++) {
             CHECK(android_stream_read_request(2, 700000, 1, -1) == 0); test_clock_poll(ua, 100); output = android_stream_pull(2, &size);
         }
         CHECK(output && output[0] == 1 && stream_read_be(output + 8, 8) == 800000 + frame * 40000); u_free(output);
     }
+    /* Пустой pull не отменяет WAIT: уже запрошенный кадр приходит после появления данных. */
+    CHECK(android_stream_read_request(2, 1000000, 1, -1) == 0); test_clock_run_for(ua, 1000);
+    CHECK(!android_stream_pull(2, &size));
+    CHECK(android_stream_feed(1, 1, 1040000, 0, video, sizeof(video), &next_format) == 1);
+    test_clock_run_for(ua, 1000);
+    output = android_stream_pull(2, &size);
+    CHECK(output && output[0] == 1 && stream_read_be(output + 8, 8) == 1040000); u_free(output);
+    /* Отмена WAIT сохраняет следующий кадр, а запрос только аудио не забирает видео. */
+    CHECK(android_stream_read_request(2, 1040000, 1, -1) == 0); test_clock_run_for(ua, 1000);
+    CHECK(!android_stream_pull(2, &size)); CHECK(android_stream_read_request(2, 1040000, 0, -1) == 0);
+    CHECK(android_stream_feed(1, 1, 1080000, 0, video, sizeof(video), &next_format) == 1);
+    CHECK(feed(1, 0, 220000, 0, (const uint8_t *)"ops", 3) == 1); test_clock_run_for(ua, 1000);
+    CHECK(!android_stream_pull(2, &size));
+    CHECK(android_stream_read_request(2, 1040000, 0, 80000) == 0); test_clock_run_for(ua, 1000);
+    output = android_stream_pull(2, &size);
+    CHECK(output && output[0] == 2 && stream_read_be(output + 8, 8) == 220000); u_free(output);
+    CHECK(android_stream_read_request(2, 1040000, 1, -1) == 0); test_clock_run_for(ua, 1000);
+    output = android_stream_pull(2, &size);
+    CHECK(output && output[0] == 1 && stream_read_be(output + 8, 8) == 1080000); u_free(output);
+    DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS demand regression: one reply, WAIT survives empty pull, cancellation and audio-only preserve video");
+    debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
     android_stream_stop(1); u_free(wait_event(2, 5, &size)); android_stream_stop(2); u_free(wait_event(1, 5, &size));
     while ((output = android_stream_pull(2, &size))) u_free(output);
     CHECK(!strcmp(source_snapshot, "[]"));