diff --git a/tests/Makefile.am b/tests/Makefile.am index c7e21590..694f9de7 100644 --- a/tests/Makefile.am +++ b/tests/Makefile.am @@ -130,6 +130,7 @@ check_PROGRAMS += \ test_android_stream \ test_stream_service \ test_stream_member \ + test_stream_publisher \ test_u_async_comprehensive \ test_nat_detection \ test_audio_compressor \ @@ -645,6 +646,13 @@ test_stream_member_CFLAGS = $(AM_CFLAGS) -DSTREAM_MEMBER_TEST_ONLY test_stream_member_LDADD = $(test_stream_service_LDADD) test_stream_member_LDFLAGS = $(test_stream_service_LDFLAGS) +test_stream_publisher_SOURCES = test_stream_publisher.c $(test_clock_sources) +test_stream_publisher_LDADD = $(LIBUTUN) $(COMMON_LIBS) +test_stream_publisher_LDFLAGS = $(test_clock_ldflags) +if OS_LINUX +test_stream_publisher_LDFLAGS += -Wl,--wrap=uasync_set_timeout +endif + test_intensive_memory_pool_SOURCES = test_intensive_memory_pool.c test_intensive_memory_pool_CFLAGS = -I$(top_srcdir)/src -I$(top_srcdir)/lib test_intensive_memory_pool_LDADD = $(COMMON_LIBS) diff --git a/tests/test_stream_publisher.c b/tests/test_stream_publisher.c new file mode 100644 index 00000000..71ef0baf --- /dev/null +++ b/tests/test_stream_publisher.c @@ -0,0 +1,197 @@ +/* Настоящие publisher/RS/uAV2: границы Opus, продвижение видео, OPENING и FIN без зависшей головы очереди. */ +#include +#include +#include + +#include "../lib/debug_config.h" +#include "../lib/mem.h" +#include "../lib/opus_codec.h" +#include "../lib/libopus/include/opus.h" +#include "../src/utun_instance.h" +#include "../src/chat/chat_core.h" +#include "../src/chat/chat_setting.h" +#include "../src/stream/stream.h" +#include "../src/stream/stream_publisher.h" +#include "test_clock.h" + +static const char *current_case; +#define CHECK(x) do { if (!(x)) { \ + DEBUG_ERROR(DEBUG_CATEGORY_DEBUG, "publisher test case=%s line=%d: %s", current_case, __LINE__, #x); exit(1); \ +} } while (0) + +struct scenario { + const char *name; + unsigned n, r, lengths[40], count; + unsigned video_bytes, opening, stop, second_video, pframe, batches; + uint64_t dropped; /* Маска ожидаемо отброшенных записей; не вычисляется из реализации budget. */ +}; +struct received { + const struct scenario *scenario; + opus_codec_decoder_t *decoder; + unsigned audio, video_bytes, batches, first_video, first_audio, next_audio; + size_t capacity; +}; + +#ifdef __linux__ +static struct stream_publisher *no_media_timer; +void *__real_uasync_set_timeout(struct UASYNC *ua, int timeout_tb, void *arg, timeout_callback_t callback, const char *name); + +/* После единственного крупного Opus короткий P-кадр должен уйти в том же actor, без media timer на 5мс. */ +void *__wrap_uasync_set_timeout(struct UASYNC *ua, int timeout_tb, void *arg, timeout_callback_t callback, const char *name) { + CHECK(arg != no_media_timer || !no_media_timer || !name || strcmp(name, "publisher_media")); + return __real_uasync_set_timeout(ua, timeout_tb, arg, callback, name); +} +#endif + +/* Проверка записей после настоящей RS-публикации: порядок, Opus-декодирование и непрерывность видео. */ +static void receive(struct UTUN_INSTANCE *inst, const struct stream_event *event, void *arg) { + (void)inst; + if (event->type != STREAM_EVENT_DATA) return; + struct received *received = arg; + const struct scenario *s = received->scenario; + struct stream_media_block block; + CHECK(event->length <= received->capacity); + CHECK(stream_media_parse(event->data, event->length, &block) == STREAM_OK); + CHECK(block.audio_count <= 8); + received->batches++; + if (block.audio_count && !received->first_audio) received->first_audio = received->batches; + if (block.video_count && !received->first_video) received->first_video = received->batches; + for (unsigned i = 0; i < block.audio_count; i++) { + unsigned index = received->next_audio; + while (index < s->count && (s->dropped & (UINT64_C(1) << index))) index++; + CHECK(index < s->count); + CHECK(block.audio[i].pts_us == 1000000u + index * 20000u); + CHECK(block.audio[i].length == s->lengths[index]); + CHECK(opus_packet_get_nb_samples(block.audio[i].data, block.audio[i].length, 48000) == 960); + int16_t pcm[1920]; + CHECK(opus_codec_decode(received->decoder, block.audio[i].data, block.audio[i].length, pcm, 960) == 960); + received->next_audio = index + 1; received->audio++; + } + for (unsigned i = 0; i < block.video_count; i++) { + CHECK(block.video[i].offset == received->video_bytes); + CHECK(block.video[i].frame_size == s->video_bytes && block.video[i].pts_us == 1000000); + CHECK(block.video[i].frame_id == (s->pframe ? 2 : 1)); + CHECK(block.video[i].flags == (s->pframe ? 0 : STREAM_MEDIA_KEY)); + for (unsigned j = 0; j < block.video[i].length; j++) { + unsigned offset = block.video[i].offset + j; + CHECK(block.video[i].data[j] == (offset < 4 ? 0 : (uint8_t)(offset * 13 + 11))); + } + received->video_bytes += block.video[i].length; + } + DEBUG_DEBUG(DEBUG_CATEGORY_DEBUG, "publisher receive case=%s batch=%u bytes=%zu audio=%u video=%u video_total=%u", + s->name, received->batches, event->length, block.audio_count, block.video_count, received->video_bytes); +} + +/* Независимое ядро/канал для каждой FEC-конфигурации; после destroy восстанавливается счётчик allocations. */ +static void check_scenario(const struct scenario *s) { + current_case = s->name; + size_t allocated = u_get_allocated_count(); + test_clock_init(100); + struct UASYNC *ua = uasync_create(); CHECK(ua); + struct UTUN_INSTANCE *inst = utun_instance_create_from_str(ua, + "[global]\nauto_sockets=no\n[chatserver]\ngroup_autoconnect=0\nstorage_autoload=0\n" + "member_files_enabled=0\nlocal_broadcasts_enabled=0\n"); + CHECK(inst && !utun_core_start(inst) && !chat_service_start(inst)); + chat_core_create_channel_auto(inst, s->name); + sqlite3_stmt *statement = NULL; + CHECK(sqlite3_prepare_v2(chat_core_get_db(inst), "SELECT channel_id FROM channels LIMIT 1", -1, &statement, NULL) == SQLITE_OK); + CHECK(sqlite3_step(statement) == SQLITE_ROW); + uint64_t group = strtoull((const char *)sqlite3_column_text(statement, 0), NULL, 10); + sqlite3_finalize(statement); CHECK(group); + char value[16]; snprintf(value, sizeof(value), "%u", s->n); + CHECK(!chat_setting_set(inst, "stream_subblocks", value)); + snprintf(value, sizeof(value), "%u", s->r); + CHECK(!chat_setting_set(inst, "stream_rs_redundant", value)); + struct stream_descriptor descriptor = {.group_id = group, .source_node_id = inst->node_id, + .source_epoch = 1, .n = s->n, .r = s->r, .block_size = STREAM_RS_DEFAULT_SIZE}; + struct received received = {.scenario = s, .capacity = stream_input_size(&descriptor)}; + uasync_mark_running(ua); + struct stream_publisher *publisher = NULL; + CHECK(!stream_publisher_create(&publisher, inst)); + struct stream_media_format format = {64, 64, 0, STREAM_MEDIA_H264, STREAM_MEDIA_OPUS_STEREO, 1}; + CHECK(!stream_publisher_open(publisher, 1, group, &format)); + struct stream_publisher_status status; + if (!s->opening) { + test_clock_run_for(ua, 10); + CHECK(!stream_publisher_snapshot(publisher, 1, &status) && status.state == STREAM_PUBLISHER_LIVE); + } + opus_codec_encoder_t *encoder = opus_codec_encoder_create(48000, 2); CHECK(encoder); + received.decoder = opus_codec_decoder_create(48000, 2); CHECK(received.decoder); + CHECK(!opus_codec_encoder_application_set(encoder, OPUS_CODEC_APP_AUDIO)); + CHECK(!opus_codec_encoder_bitrate_set(encoder, 128000)); + int16_t pcm[1920] = {0}; uint8_t original[1275], packet[1275], video[6000] = {0}; + for (unsigned i = 4; i < sizeof(video); i++) video[i] = (uint8_t)(i * 13 + 11); + int encoded = opus_codec_encode(encoder, pcm, 960, original, sizeof(original)); CHECK(encoded > 0); + if (s->pframe) { + CHECK(stream_publisher_feed(publisher, 1, 1, 960000, 1, video, 5, &format) == 1); + test_clock_run_for(ua, 300); +#ifdef __linux__ + no_media_timer = publisher; +#endif + } + stream_set_event_callback(inst, receive, &received); + for (unsigned i = 0; i < s->count; i++) { + CHECK(s->lengths[i] >= (unsigned)encoded && s->lengths[i] <= sizeof(packet)); + memcpy(packet, original, encoded); + CHECK(opus_packet_pad(packet, encoded, s->lengths[i]) == OPUS_OK); + CHECK(stream_publisher_feed(publisher, 1, 0, 1000000u + i * 20000u, 0, packet, s->lengths[i], NULL) == 1); + } + CHECK(s->video_bytes <= sizeof(video)); + if (s->video_bytes) CHECK(stream_publisher_feed(publisher, 1, 1, 1000000, !s->pframe, video, s->video_bytes, &format) == 1); + if (s->stop) stream_publisher_stop(publisher, 1); + test_clock_run_for(ua, 2000); /* До TTL очереди: потеря или задержка до expiry не считается успехом. */ + CHECK(!stream_publisher_snapshot(publisher, 1, &status)); + unsigned expected_audio = s->count; + for (unsigned i = 0; i < s->count; i++) if (s->dropped & (UINT64_C(1) << i)) expected_audio--; + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "publisher case=%s audio=%u/%u video=%u/%u batches=%u first_audio=%u first_video=%u state=%d", + s->name, received.audio, expected_audio, received.video_bytes, s->video_bytes, + received.batches, received.first_audio, received.first_video, status.state); + CHECK(received.audio == expected_audio && received.video_bytes == s->video_bytes); + if (s->second_video) CHECK(received.first_audio == 1 && received.first_video == 2); + if (s->batches) CHECK(received.batches == s->batches); + CHECK(status.state == (s->stop ? STREAM_PUBLISHER_DRAINING : STREAM_PUBLISHER_LIVE)); +#ifdef __linux__ + no_media_timer = NULL; +#endif + stream_set_event_callback(inst, NULL, NULL); stream_publisher_destroy(publisher); + opus_codec_encoder_destroy(encoder); opus_codec_decoder_destroy(received.decoder); + uasync_mark_stopped(ua); utun_instance_destroy(inst); uasync_destroy(ua, 0); test_clock_finish(); + CHECK(u_get_allocated_count() == allocated); +} + +int main(void) { + debug_config_init(); debug_set_level(DEBUG_LEVEL_ERROR); + debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_WARN); + const struct scenario cases[] = { + {"old-boundary", 1, 0, {1044}, 1, .batches = 1}, + {"new-boundary", 1, 0, {1045}, 1, .batches = 1}, + {"exact-capacity", 1, 0, {1077}, 1, .batches = 1}, + {"unfit-next-fit", 1, 0, {1078, 320}, 2, .dropped = 1}, + {"unfit-opening", 1, 0, {1275, 320}, 2, .opening = 1, .dropped = 1}, + {"unfit-middle", 1, 0, {320, 1078, 320}, 3, .dropped = 2}, + {"unfit-consecutive", 1, 0, {1078, 1275, 320}, 3, .dropped = 3}, + {"two-audio-exact", 1, 0, {530, 531}, 2, .batches = 1}, + {"two-audio-split", 1, 0, {600, 600}, 2, .batches = 2}, + {"mixed-one-byte", 1, 0, {1044}, 1, .video_bytes = 6000}, + {"mixed-audio-only", 1, 0, {1045}, 1, .video_bytes = 6000, .second_video = 1}, + {"mixed-exact", 1, 0, {1077}, 1, .video_bytes = 6000, .second_video = 1}, + {"large-audio-backlog", 1, 0, + {1077,1077,1077,1077,1077,1077,1077,1077,1077,1077,1077,1077,1077,1077,1077,1077, + 1077,1077,1077,1077,1077,1077,1077,1077,1077,1077,1077,1077,1077,1077,1077,1077}, + 32, .video_bytes = 6000, .second_video = 1}, + {"mixed-small-video", 1, 0, {1077, 1077}, 2, .video_bytes = 5, .second_video = 1}, + {"stop-with-queues", 1, 0, {1077, 320}, 2, .video_bytes = 6000, .stop = 1, .second_video = 1}, + {"no-fit-stop", 1, 0, {1275}, 1, .stop = 1, .dropped = 1}, + {"opening-large", 1, 0, {1077}, 1, .video_bytes = 6000, .opening = 1, .second_video = 1}, + {"redundant-one-data", 2, 1, {1077}, 1, .video_bytes = 6000, .second_video = 1}, + {"two-data-large", 2, 0, {1275}, 1, .video_bytes = 6000}, + {"default-large", 8, 0, {1275, 1275}, 2, .video_bytes = 6000, .batches = 1}, + {"video-only", 1, 0, {0}, 0, .video_bytes = 6000}, + {"pframe-coalescing", 1, 0, {1077}, 1, .video_bytes = 5, .second_video = 1, .pframe = 1}, + {"audio-eight-limit", 8, 0, {320,320,320,320,320,320,320,320,320,320}, 10, .batches = 2} + }; + for (unsigned i = 0; i < sizeof(cases) / sizeof(cases[0]); i++) check_scenario(&cases[i]); + DEBUG_INFO(DEBUG_CATEGORY_DEBUG, "PASS stream_publisher: %zu cases, real RS/Opus, video progress, OPENING/FIN, no tracked leaks", + sizeof(cases) / sizeof(cases[0])); + return 0; +}