jeka
/
utun2
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

Gate media cursor revision changes on a complete keyframe

master
evgeny 5 days ago
parent
afa8fd307a
commit
d209ab94ba
2 changed files with 101 additions and 17 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 28
      src/stream/stream_media_buffer.c
  2. 90
      tests/test_stream_media.c

28
src/stream/stream_media_buffer.c
Unescape View File

@ -30,8 +30,8 @@ struct cursor_track { uint64_t seq; unsigned entry; int ended; };
struct stream_media_cursor { struct stream_media_cursor {
struct stream_media_store *store; struct stream_media_store *store;
struct cursor_track video, audio; struct cursor_track video, audio;
uint32_t generation; uint32_t generation, video_revision;
int initialized, key_delivered; int initialized;
unsigned tracks; unsigned tracks;
int64_t start_us; int64_t start_us;
uint8_t *scratch[2], *frame; uint8_t *scratch[2], *frame;
@ -257,7 +257,8 @@ void stream_media_cursor_destroy(struct stream_media_cursor *c) {
void stream_media_cursor_reset(struct stream_media_cursor *c) { void stream_media_cursor_reset(struct stream_media_cursor *c) {
if (!c) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media cursor reset: null cursor"); return; } if (!c) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media cursor reset: null cursor"); return; }
c->initialized = c->key_delivered = 0; c->pending_used = 0; c->initialized = 0; c->video_revision = 0; c->pending_used = 0;
c->pending = (struct stream_media_video){0}; c->pending_format = (struct stream_media_format){0};
c->scratch_valid[0] = c->scratch_valid[1] = 0; c->scratch_valid[0] = c->scratch_valid[1] = 0;
c->video = (struct cursor_track){0}; c->audio = (struct cursor_track){0}; c->video = (struct cursor_track){0}; c->audio = (struct cursor_track){0};
c->audio.ended = !(c->tracks & STREAM_MEDIA_TRACK_AUDIO); c->audio.ended = !(c->tracks & STREAM_MEDIA_TRACK_AUDIO);
@ -358,10 +359,19 @@ static int cursor_video(struct stream_media_cursor *c, struct stream_media_packe
int result = cursor_peek(c, &c->video, 1, &b, &pts); int result = cursor_peek(c, &c->video, 1, &b, &pts);
if (result != STREAM_OK) return result; if (result != STREAM_OK) return result;
const struct stream_media_video *v = &b.video[c->video.entry++]; const struct stream_media_video *v = &b.video[c->video.entry++];
if (!c->key_delivered && !c->pending.frame_size if (b.format.revision != c->pending_format.revision) {
DEBUG_DEBUG(DEBUG_CATEGORY_STREAM, "media cursor revision source=%llx seq=%llu revision=%u->%u pending=%u; require KEY",
(unsigned long long)c->store->descriptor.source_node_id, (unsigned long long)c->video.seq,
c->pending_format.revision, b.format.revision, c->pending_used);
c->pending = (struct stream_media_video){0}; c->pending_used = 0; c->pending_format = b.format;
}
/* Revision относится к уже выданному видео этого reader, а не к текущему live store. */
if (b.format.revision != c->video_revision && !(v->flags & STREAM_MEDIA_KEY)) continue;
if (!c->pending.frame_size && v->offset) continue;
if (!c->video_revision && !c->pending.frame_size
&& (v->offset || !(v->flags & STREAM_MEDIA_KEY) || pts != c->start_us)) continue; && (v->offset || !(v->flags & STREAM_MEDIA_KEY) || pts != c->start_us)) continue;
if (!v->offset) { if (!v->offset) {
if (!c->key_delivered && (!(v->flags & STREAM_MEDIA_KEY) || pts != c->start_us)) continue; if (!c->video_revision && (!(v->flags & STREAM_MEDIA_KEY) || pts != c->start_us)) continue;
if (v->frame_size > c->frame_capacity) { if (v->frame_size > c->frame_capacity) {
uint8_t *frame = u_realloc(c->frame, v->frame_size); uint8_t *frame = u_realloc(c->frame, v->frame_size);
if (!frame) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media cursor frame allocation failed bytes=%u", v->frame_size); return STREAM_ERR_MEMORY; } if (!frame) { DEBUG_ERROR(DEBUG_CATEGORY_STREAM, "media cursor frame allocation failed bytes=%u", v->frame_size); return STREAM_ERR_MEMORY; }
@ -370,7 +380,7 @@ static int cursor_video(struct stream_media_cursor *c, struct stream_media_packe
c->pending = *v; c->pending.data = NULL; c->pending_used = 0; c->pending_format = b.format; c->pending = *v; c->pending.data = NULL; c->pending_used = 0; c->pending_format = b.format;
} }
if (!c->pending.frame_size || v->frame_id != c->pending.frame_id || v->offset != c->pending_used if (!c->pending.frame_size || v->frame_id != c->pending.frame_id || v->offset != c->pending_used
|| v->pts_us != c->pending.pts_us || v->frame_size != c->pending.frame_size) { || v->pts_us != c->pending.pts_us || v->frame_size != c->pending.frame_size || v->flags != c->pending.flags) {
DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media cursor fragment conflict seq=%llu frame=%llu offset=%u used=%u", DEBUG_WARN(DEBUG_CATEGORY_STREAM, "media cursor fragment conflict seq=%llu frame=%llu offset=%u used=%u",
(unsigned long long)c->video.seq, (unsigned long long)v->frame_id, v->offset, c->pending_used); (unsigned long long)c->video.seq, (unsigned long long)v->frame_id, v->offset, c->pending_used);
return STREAM_GAP; return STREAM_GAP;
@ -379,7 +389,11 @@ static int cursor_video(struct stream_media_cursor *c, struct stream_media_packe
if (c->pending_used != c->pending.frame_size) continue; if (c->pending_used != c->pending.frame_size) continue;
struct stream_media_packet out = {.kind = STREAM_MEDIA_READ_VIDEO, .format = c->pending_format, .video = c->pending}; struct stream_media_packet out = {.kind = STREAM_MEDIA_READ_VIDEO, .format = c->pending_format, .video = c->pending};
out.video.offset = 0; out.video.length = out.video.frame_size; out.video.data = c->frame; out.video.offset = 0; out.video.length = out.video.frame_size; out.video.data = c->frame;
c->pending_used = 0; c->pending.frame_size = 0; c->key_delivered = 1; *packet = out; if (c->video_revision != b.format.revision) DEBUG_INFO(DEBUG_CATEGORY_STREAM,
"media cursor KEY ready source=%llx revision=%u frame=%llu pts=%llu bytes=%u",
(unsigned long long)c->store->descriptor.source_node_id, b.format.revision,
(unsigned long long)v->frame_id, (unsigned long long)v->pts_us, v->frame_size);
c->pending_used = 0; c->pending.frame_size = 0; c->video_revision = b.format.revision; *packet = out;
return STREAM_OK; return STREAM_OK;
} }
return STREAM_MORE; return STREAM_MORE;

90
tests/test_stream_media.c
Unescape View File

@ -260,26 +260,94 @@ static int media_fixture_feed(struct media_fixture *f, const struct stream_media
return stream_media_store_feed(f->store, f->seq++, decoded, decoded_size); return stream_media_store_feed(f->store, f->seq++, decoded, decoded_size);
} }
/* Диагностика расхождения store/cursor; отдельный запуск --revision-transition до согласования контракта. */ /* Новая revision начинается с целого KEY; аудио и отстающий reader сохраняют собственные позиции. */
static void test_revision_transition(void) { static void test_revision_transition(void) {
struct media_fixture f; media_fixture_open(&f); struct media_fixture f; media_fixture_open(&f);
struct stream_media_block b = {.format = {1280, 720, 90, STREAM_MEDIA_H264, 0, 1}, .video_count = 1}; struct stream_media_block b = {.format = {1280, 720, 90, STREAM_MEDIA_H264, STREAM_MEDIA_OPUS_STEREO, 1},
.video_count = 1, .audio_count = 1};
b.video[0] = (struct stream_media_video){1, 1000000, 12, 0, 12, STREAM_MEDIA_KEY, expected_video}; b.video[0] = (struct stream_media_video){1, 1000000, 12, 0, 12, STREAM_MEDIA_KEY, expected_video};
b.audio[0] = (struct stream_media_audio){1000000, 3, 960, (const uint8_t *)"ops"};
CHECK(media_fixture_feed(&f, &b) == STREAM_OK); CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
struct stream_media_packet packet; struct stream_media_packet packet; struct stream_media_status status; struct stream_media_cursor *slow = NULL;
CHECK(stream_media_cursor_create(&slow, f.store, STREAM_MEDIA_TRACK_VIDEO) == STREAM_OK);
CHECK(stream_media_cursor_status(slow, &status) == STREAM_OK && status.initialized);
CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_OK && packet.format.revision == 1); CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_OK && packet.format.revision == 1);
b.audio_count = 0;
b.video[0] = (struct stream_media_video){2, 1040000, 12, 0, 6, 0, expected_video}; b.video[0] = (struct stream_media_video){2, 1040000, 12, 0, 6, 0, expected_video};
CHECK(media_fixture_feed(&f, &b) == STREAM_OK); CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
b.format = (struct stream_media_format){1920, 1080, 270, STREAM_MEDIA_H264, 0, 2}; CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_WAIT);
b.format = (struct stream_media_format){1920, 1080, 270, STREAM_MEDIA_H264, STREAM_MEDIA_OPUS_STEREO, 2};
b.video[0] = (struct stream_media_video){3, 1080000, 12, 0, 12, 0, expected_video}; b.video[0] = (struct stream_media_video){3, 1080000, 12, 0, 12, 0, expected_video};
CHECK(media_fixture_feed(&f, &b) == STREAM_OK); CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
struct stream_media_status status;
CHECK(stream_media_cursor_status(f.cursor, &status) == STREAM_OK && status.video_us == 1000000); CHECK(stream_media_cursor_status(f.cursor, &status) == STREAM_OK && status.video_us == 1000000);
int result = stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet); CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_WAIT);
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "revision transition before KEY: indexed_video_us=%lld result=%d kind=%d revision=%u frame=%llu flags=%u", CHECK(stream_media_cursor_read(f.cursor, 1000000, 0, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_AUDIO);
(long long)status.video_us, result, packet.kind, packet.format.revision, (unsigned long long)packet.video.frame_id, packet.video.flags); CHECK(packet.audio.pts_us == 1000000 && packet.format.audio_codec == STREAM_MEDIA_OPUS_STEREO);
b.video_count = 0; b.audio_count = 1; b.audio[0].pts_us = 1020000;
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
CHECK(stream_media_cursor_read(f.cursor, 1020000, 0, 0, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_AUDIO);
CHECK(packet.audio.pts_us == 1020000);
b.video_count = 1; b.audio_count = 0;
b.video[0] = (struct stream_media_video){4, 1120000, 12, 0, 6, STREAM_MEDIA_KEY, expected_video};
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_WAIT);
b.video[0].offset = 6; b.video[0].data = expected_video + 6;
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO);
CHECK(packet.format.revision == 2 && packet.video.frame_id == 4 && packet.video.flags == STREAM_MEDIA_KEY);
CHECK(packet.video.length == 12 && !memcmp(packet.video.data, expected_video, 12));
CHECK(stream_media_cursor_read(slow, -1, 1, -1, &packet) == STREAM_OK && packet.format.revision == 1);
CHECK(stream_media_cursor_read(slow, 1000000, 1, -1, &packet) == STREAM_OK && packet.format.revision == 2);
CHECK(packet.video.frame_id == 4);
/* Полный KEY новой конфигурации выдаётся сразу, без таймера и искусственной задержки. */
b.format.revision = 3; b.video[0] = (struct stream_media_video){5, 1160000, 12, 0, 12, STREAM_MEDIA_KEY, expected_video};
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
CHECK(stream_media_cursor_read(f.cursor, 1120000, 1, -1, &packet) == STREAM_OK && packet.format.revision == 3);
CHECK(packet.video.frame_id == 5 && packet.video.flags == STREAM_MEDIA_KEY);
stream_media_cursor_destroy(slow);
media_fixture_close(&f); media_fixture_close(&f);
CHECK(result == STREAM_WAIT); DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS cursor revisions: missing/fragmented KEY, immediate complete KEY, independent audio and history");
}
/* Пропущенный блок, нарушенные фрагменты и вытеснение запрещают P-кадры до нового целого KEY. */
static void test_cursor_key_recovery(void) {
for (unsigned failure = 0; failure < 3; failure++) {
struct media_fixture f; media_fixture_open(&f);
struct stream_media_block b = {.format = {1280, 720, 0, STREAM_MEDIA_H264, 0, 1}, .video_count = 1};
b.video[0] = (struct stream_media_video){1, 1000000, 12, 0, 12, STREAM_MEDIA_KEY, expected_video};
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
struct stream_media_packet packet;
CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_OK);
b.video[0] = (struct stream_media_video){2, 1040000, 12, 0, 6, 0, expected_video};
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
CHECK(stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet) == STREAM_WAIT);
if (failure == 0) { stream_media_store_gap(f.store, f.seq, f.seq); f.seq++; }
else if (failure == 1) {
b.video[0].offset = 5; b.video[0].data = expected_video + 5;
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
} else {
for (unsigned i = 0; i < 48; i++) {
b.video[0] = (struct stream_media_video){3 + i, 1080000 + i * 40000, 12, 0, 12, 0, expected_video};
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
}
}
int result = stream_media_cursor_read(f.cursor, 1000000, 1, -1, &packet);
CHECK(result == (failure == 2 ? STREAM_EXPIRED : STREAM_GAP));
stream_media_cursor_reset(f.cursor);
b.video[0] = (struct stream_media_video){100, 4000000, 12, 0, 12, 0, expected_video};
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_WAIT);
b.video[0] = (struct stream_media_video){101, 4040000, 12, 0, 6, STREAM_MEDIA_KEY, expected_video};
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_WAIT);
b.video[0].offset = 6; b.video[0].data = expected_video + 6;
CHECK(media_fixture_feed(&f, &b) == STREAM_OK);
CHECK(stream_media_cursor_read(f.cursor, -1, 1, -1, &packet) == STREAM_OK && packet.kind == STREAM_MEDIA_READ_VIDEO);
CHECK(packet.video.frame_id == 101 && packet.video.flags == STREAM_MEDIA_KEY && packet.video.length == 12);
CHECK(!memcmp(packet.video.data, expected_video, 12));
media_fixture_close(&f);
}
DEBUG_INFO(DEBUG_CATEGORY_STREAM, "PASS cursor recovery: GAP, fragment discontinuity and expiry resume only from a complete KEY");
} }
int main(int argc, char **argv) { int main(int argc, char **argv) {
@ -290,10 +358,12 @@ int main(int argc, char **argv) {
for (unsigned i = 0; i < sizeof(expected_video); i++) expected_video[i] = (uint8_t)(i * 31 + 7); for (unsigned i = 0; i < sizeof(expected_video); i++) expected_video[i] = (uint8_t)(i * 31 + 7);
if (argc == 2 && !strcmp(argv[1], "--revision-transition")) { if (argc == 2 && !strcmp(argv[1], "--revision-transition")) {
debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_DEBUG); debug_set_level(DEBUG_LEVEL_ERROR); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_DEBUG);
test_revision_transition(); test_revision_transition(); test_cursor_key_recovery();
} else { } else {
CHECK(argc == 1); CHECK(argc == 1);
test_media(); test_batches(); test_revisions(); test_stereo(); test_codecs(); test_media(); test_batches(); test_revisions(); test_stereo(); test_codecs();
debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);
test_revision_transition(); test_cursor_key_recovery();
} }
CHECK(u_get_allocated_count() == before); CHECK(u_get_allocated_count() == before);
debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO); debug_set_category_level(DEBUG_CATEGORY_STREAM, DEBUG_LEVEL_INFO);

Write Preview
Loading…
Cancel
Save