From ea5d1fcd4561fbb654ac831f513af31310ea4fba Mon Sep 17 00:00:00 2001 From: evgeny Date: Tue, 20 Oct 2026 18:45:49 +0200 Subject: [PATCH] Consume repeated viewer demand when delivering its reply --- src/stream/stream_viewer.c | 17 ++++++++++++++--- src/stream/stream_viewer.h | 3 ++- tests/test_android_stream.c | 30 +++++++++++++++++++++++++++++- 3 files changed, 45 insertions(+), 5 deletions(-) diff --git a/src/stream/stream_viewer.c b/src/stream/stream_viewer.c index 72ba7ecb..76d58bdd 100644 --- a/src/stream/stream_viewer.c +++ b/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; } diff --git a/src/stream/stream_viewer.h b/src/stream/stream_viewer.h index ca44e584..6daf9b6f 100644 --- a/src/stream/stream_viewer.h +++ b/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. */ diff --git a/tests/test_android_stream.c b/tests/test_android_stream.c index e77679e9..cd910c4a 100644 --- a/tests/test_android_stream.c +++ b/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, "[]"));