From 599c1492eeffdfe9509bbe5f5c3126698b1a5306 Mon Sep 17 00:00:00 2001 From: hl-valdemar Date: Sat, 17 Oct 2026 00:56:20 +0200 Subject: [PATCH] perf: retain Cast codecs across sequential VOD segments --- Checks/cast-media.cpp | 85 ++++++++++++++++++++++++ Makefile | 1 + Sources/Core/cast_media.cpp | 129 ++++++++++++++++++++++++++---------- TODO.md | 3 +- 4 files changed, 182 insertions(+), 36 deletions(-) diff --git a/Checks/cast-media.cpp b/Checks/cast-media.cpp index 9ef92a7..8caae92 100644 --- a/Checks/cast-media.cpp +++ b/Checks/cast-media.cpp @@ -14,6 +14,7 @@ extern "C" { #include #include #include +#include #include #include #include @@ -673,6 +674,84 @@ static void remuxWork(const std::filesystem::path &directory) { cast_media_destroy(media); } } +// Exercise the HTTP relay, not an isolated codec: delayed video, resampled AAC, +// independent segment decoding, cache eviction, and random-to-sequential seeks. +static void codecWork(const std::filesystem::path &directory) { + const auto source = (directory / "codec.mp4").string(), audio = (directory / "codec.wav").string(); + const char *args[] = {"ffmpeg", "-hide_banner", "-loglevel", "error", "-y", + "-f", "lavfi", "-i", "testsrc2=size=1280x720:rate=30:duration=61", + "-f", "lavfi", "-i", "sine=frequency=440:sample_rate=44100:duration=61", + "-c:v", "libx264", "-preset", "ultrafast", "-bf", "2", "-g", "210", + "-an", source.c_str(), "-vn", "-c:a", "pcm_s16le", audio.c_str(), nullptr}; + pid_t child = 0; + assert(posix_spawnp(&child, args[0], nullptr, nullptr, const_cast(args), environ) == 0); + int status = 0; assert(waitpid(child, &status, 0) == child); + assert(WIFEXITED(status) && WEXITSTATUS(status) == 0); + Upstream upstream(readFile(source), readFile(audio)); upstream.start(); + const auto videoURL = upstream.url("/source.ts"), audioURL = upstream.url("/audio.wav"); + std::string url; double duration = 0; + const auto began = std::chrono::steady_clock::now(); + auto *media = start({videoURL.c_str(), "X-Video: video-secret\r\n", audioURL.c_str(), + "X-Audio: audio-secret\r\n", false, false, false, 0}, url, duration); + const auto prepared = std::chrono::steady_clock::now(); + const auto base = url.substr(0, url.rfind('/') + 1), output = (directory / "codec-output.ts").string(); + std::map> spans; + double sequentialMS = 0, randomMS = 0, sequentialCPU = 0; + for (int id : {0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 2, 3, 0, 1, 10}) { + const auto before = std::chrono::steady_clock::now(); + const auto cpuBefore = std::clock(); + const auto response = request(base + "segment/" + std::to_string(id) + ".ts"); + const auto elapsed = std::chrono::duration(std::chrono::steady_clock::now() - before).count(); + const double cpuMS = 1000.0 * (std::clock() - cpuBefore) / CLOCKS_PER_SEC; + if (spans.size() < 10) { sequentialMS += elapsed; sequentialCPU += cpuMS; } + else randomMS += elapsed; + assert(response.status == 200); writeFile(output, response.bytes); + const auto rows = samples(output, 1); + int videos = 0; + for (const auto &[key, value] : rows) { + assert(key.second >= id * 6 * 90000 && key.second < std::min(61, (id + 1) * 6) * 90000); + if (key.first) ++videos; + } + assert(videos == (id == 10 ? 30 : 180)); + auto span = audioSpan(response, output); + if (spans.contains(id - 1)) assert(std::abs(span.first - spans.at(id - 1).second) < 1.0 / 48000); + if (spans.contains(id + 1)) assert(std::abs(spans.at(id + 1).first - span.second) < 1.0 / 48000); + spans[id] = span; + assert(request(base + "segment/" + std::to_string(id) + ".ts").bytes == response.bytes); + } + assert(upstream.foreignRequests == 0); + cast_media_destroy(media); + std::cout << "codec-work prepare_ms=" << std::chrono::duration(prepared - began).count() + << " sequential_60s_ms=" << sequentialMS << " sequential_cpu_ms=" << sequentialCPU + << " seek_sequence_ms=" << randomMS << std::endl; + // A long video gap must not turn bounded encoder lookahead into HTTP 503, + // and resuming video after audio-only segments still needs an independent IDR. + const auto sparsePath = (directory / "sparse.mp4").string(); + const char *sparseArgs[] = {"ffmpeg", "-hide_banner", "-loglevel", "error", "-y", + "-i", source.c_str(), "-t", "25", "-vf", "select='lt(t,3)+gte(t,18)'", + "-fps_mode", "vfr", "-c:v", "libx264", "-preset", "ultrafast", "-bf", "2", + "-an", sparsePath.c_str(), nullptr}; + assert(posix_spawnp(&child, sparseArgs[0], nullptr, nullptr, const_cast(sparseArgs), environ) == 0); + assert(waitpid(child, &status, 0) == child && WIFEXITED(status) && WEXITSTATUS(status) == 0); + Upstream sparse(readFile(sparsePath), readFile(audio)); sparse.start(); + const auto sparseVideo = sparse.url("/source.ts"), sparseAudio = sparse.url("/audio.wav"); + media = start({sparseVideo.c_str(), "X-Video: video-secret\r\n", sparseAudio.c_str(), + "X-Audio: audio-secret\r\n", false, false, false, 0}, url, duration); + const auto sparseBase = url.substr(0, url.rfind('/') + 1); + double audioEnd = NAN; + for (int id = 0; id < 5; ++id) { + const auto response = request(sparseBase + "segment/" + std::to_string(id) + ".ts"); + const auto span = audioSpan(response, output); + if (id) assert(std::abs(span.first - audioEnd) < 1.0 / 48000); + audioEnd = span.second; + if (id == 0 || id >= 3) { + const auto rows = samples(output, 1); + const auto videos = std::count_if(rows.begin(), rows.end(), [](const auto &row) { return row.first.first == 1; }); + assert(videos == (id == 0 ? 90 : id == 3 ? 180 : 30)); + } + } + cast_media_destroy(media); +} int main(int argc, char **argv) { bool liveCheck = false, seekCheck = false, hlsCheck = false, remuxCheck = false; for (int argument = 1; argument < argc; ++argument) { @@ -680,6 +759,12 @@ int main(int argc, char **argv) { if (std::string(argv[argument]) == "--seek-work") seekCheck = true; if (std::string(argv[argument]) == "--hls-work") hlsCheck = true; if (std::string(argv[argument]) == "--remux-work") remuxCheck = true; + if (std::string(argv[argument]) == "--codec-work") { + char temp[] = "/tmp/terebi-codec-check.XXXXXX"; assert(mkdtemp(temp)); + codecWork(temp); std::filesystem::remove_all(temp); + std::cout << "cast-media: sequential codec reuse, independent IDRs, AAC continuity and random seeks passed\n"; + return 0; + } } char temp[] = "/tmp/terebi-cast-check.XXXXXX"; assert(mkdtemp(temp)); std::filesystem::path directory(temp); if (remuxCheck) { diff --git a/Makefile b/Makefile index 99cf03a..93ca69a 100644 --- a/Makefile +++ b/Makefile @@ -249,6 +249,7 @@ check-cast: $(BUILD_DIR)/cast-session-check $(BUILD_DIR)/cast-media-check $(BUIL $(BUILD_DIR)/cast-session-check $(BUILD_DIR)/cast-media-check --live $(BUILD_DIR)/cast-media-check --remux-work + $(BUILD_DIR)/cast-media-check --codec-work $(BUILD_DIR)/cast-session-check --playback $(BUILD_DIR)/cast-playback-check $(BUILD_DIR)/fixtures/cache.mp4 $(BUILD_DIR)/app-startup-check.dylib: Checks/app-startup.swift $(SWIFT_SOURCES) $(OBJECTS) $(CONFIG_LIBRARY) Makefile | check-dependencies diff --git a/Sources/Core/cast_media.cpp b/Sources/Core/cast_media.cpp index 860da31..c548578 100644 --- a/Sources/Core/cast_media.cpp +++ b/Sources/Core/cast_media.cpp @@ -342,9 +342,12 @@ struct OutputTrack { int64_t audioPTS = AV_NOPTS_VALUE; bool firstVideo = true; double lastVideoTime = -INFINITY; - bool continuous; - int64_t audioKeepFrom = AV_NOPTS_VALUE, audioKeepUntil = INT64_MAX; - OutputTrack(Track &track, Mux &output, bool live) : source(track), mux(&output), continuous(live) { + bool live, drained = false; + std::deque packets; + size_t packetBytes = 0; + int64_t lastPacketPTS = AV_NOPTS_VALUE; + int64_t keepFrom = INT64_MIN, keepUntil = INT64_MAX; + OutputTrack(Track &track, Mux &output, bool live) : source(track), mux(&output), live(live) { try { if (!source.stream) return; stream = avformat_new_stream(mux->format, nullptr); require(stream != nullptr); @@ -397,7 +400,6 @@ struct OutputTrack { stream->time_base = encoder->time_base; check(avcodec_parameters_from_context(stream->codecpar, encoder)); } - firstVideo = true; } void writePacket(AVPacket *packet) { // A shared, positive TS clock origin prevents per-segment AAC priming shifts. @@ -413,10 +415,16 @@ struct OutputTrack { int result = avcodec_receive_packet(encoder, p.get()); if (result == AVERROR(EAGAIN) || result == AVERROR_EOF) break; check(result); - if (fifo && !continuous && audioKeepFrom != AV_NOPTS_VALUE && - (p->pts < audioKeepFrom || p->pts >= audioKeepUntil)) { - av_packet_unref(p.get()); - continue; + lastPacketPTS = p->pts; + if (!live) { + require(p->pts != AV_NOPTS_VALUE && p->size >= 0); + if (p->pts >= keepUntil) { + require(packetBytes + size_t(p->size) <= segmentLimit); + packetBytes += size_t(p->size); + packets.push_back(std::move(p)); p = packet(); + continue; + } + if (p->pts < keepFrom) { av_packet_unref(p.get()); continue; } } av_packet_rescale_ts(p.get(), encoder->time_base, stream->time_base); writePacket(p.get()); av_packet_unref(p.get()); @@ -463,12 +471,8 @@ struct OutputTrack { } } else { require(f->sample_rate > 0 && f->nb_samples >= 0 && f->nb_samples <= 65536 && f->ch_layout.nb_channels <= 32); - const double begin = continuous ? start : aacPreroll(start); - const double limit = continuous ? end : double(aacBoundary(end)) / 48000; - if (!continuous) { - audioKeepFrom = aacBoundary(start); - audioKeepUntil = int64_t(std::llround(end * 48000)); - } + const double begin = live ? start : aacPreroll(start); + const double limit = end; if (time + double(f->nb_samples) / f->sample_rate > begin && time < limit) { if (!resample) { check(swr_alloc_set_opts2(&resample, &encoder->ch_layout, encoder->sample_fmt, 48000, &f->ch_layout, AVSampleFormat(f->format), f->sample_rate, 0, nullptr)); @@ -479,10 +483,9 @@ struct OutputTrack { auto out = frame(); out->format = encoder->sample_fmt; out->sample_rate = 48000; check(av_channel_layout_copy(&out->ch_layout, &encoder->ch_layout)); out->nb_samples = capacity; check(av_frame_get_buffer(out.get(), 0)); int count = swr_convert(resample, out->data, capacity, const_cast(f->extended_data), f->nb_samples); check(count); - int skip = continuous && audioPTS != AV_NOPTS_VALUE ? 0 : + int skip = audioPTS != AV_NOPTS_VALUE ? 0 : std::clamp(int(std::llround((begin - time) * 48000)), 0, count); - int take = continuous ? count - skip : - std::clamp(int(std::llround((limit - time) * 48000)) - skip, 0, count - skip); + int take = count - skip; if (audioPTS == AV_NOPTS_VALUE) audioPTS = int64_t(std::llround(std::max(begin, time) * 48000)); uint8_t *data[2] = {out->data[0] + skip * 4, out->data[1] + skip * 4}; require(av_audio_fifo_size(fifo) + take <= 131072); @@ -491,13 +494,52 @@ struct OutputTrack { } source.nextFrame(); } - void finish(double end) { + void beginPackets(double start, double end) { + if (!encoder || live) return; + keepFrom = fifo ? aacBoundary(start) : int64_t(std::llround(start * 90000)); + keepUntil = fifo ? int64_t(std::llround(end * 48000)) : int64_t(std::llround(end * 90000)); + while (!packets.empty() && packets.front()->pts < keepUntil) { + auto p = std::move(packets.front()); packets.pop_front(); + packetBytes -= size_t(p->size); + if (p->pts < keepFrom) continue; + av_packet_rescale_ts(p.get(), encoder->time_base, stream->time_base); + writePacket(p.get()); + } + } + void finish(double start, double end) { if (!encoder) return; - if (continuous) { - if (fifo) audioFrames(false, end); - } else { - if (fifo) audioFrames(true); - encoded(nullptr); + if (live) { + if (fifo) audioFrames(false, end); else encoded(nullptr); + return; + } + // Neither encoder can resume after a drain. Feed just enough future + // input to receive this segment's delayed packets; retain the rest. + const int64_t target = fifo ? aacBoundary(end) - encoder->frame_size : + (std::isfinite(lastVideoTime) ? int64_t(std::llround(lastVideoTime * 90000)) : AV_NOPTS_VALUE); + if (!fifo) firstVideo = true; + while (!drained && target != AV_NOPTS_VALUE && (lastPacketPTS == AV_NOPTS_VALUE || lastPacketPTS < target)) { + if (fifo) audioFrames(false); + if (lastPacketPTS != AV_NOPTS_VALUE && lastPacketPTS >= target) break; + const double time = source.time(); + // ponytail: cap lookahead at six seconds; drain and reset on the + // next segment rather than following a sparse stream arbitrarily far. + if (!std::isfinite(time) || time >= end + segmentSeconds) { + if (fifo) { + // Drain resampler delay before padding the final AAC frame. + if (resample) { + auto tail = frame(); tail->format = encoder->sample_fmt; + tail->nb_samples = 4096; + check(av_channel_layout_copy(&tail->ch_layout, &encoder->ch_layout)); + check(av_frame_get_buffer(tail.get(), 0)); + int count = swr_convert(resample, tail->data, tail->nb_samples, nullptr, 0); check(count); + require(av_audio_fifo_write(fifo, reinterpret_cast(tail->data), count) == count); + } + audioFrames(true); + } + encoded(nullptr); drained = true; + break; + } + consume(fifo ? start : end, INFINITY); } } }; @@ -513,6 +555,7 @@ struct Pipeline { std::atomic &cancel; double duration = 0; std::vector boundaries; + double nextStart = 0; Pipeline(const Source &source, std::atomic &stop, double start) : live(source.live), cancel(stop) { videoInput = std::make_unique(stop, source.video, source.videoHeaders, source.local, source.restricted, source.live); duration = std::max(source.duration, videoInput->duration()); @@ -567,8 +610,8 @@ struct Pipeline { void armOperation() { deadline = Clock::now() + std::chrono::seconds(45); const auto limit = std::chrono::duration_cast(deadline.time_since_epoch()).count(); - videoInput->operationDeadline = limit; - if (audioInput) audioInput->operationDeadline = limit; + videoInput->operationDeadline = limit; videoInput->arm(); + if (audioInput) { audioInput->operationDeadline = limit; audioInput->arm(); } } void releaseOperation() { videoInput->operationDeadline = std::numeric_limits::max(); @@ -578,8 +621,8 @@ struct Pipeline { armOperation(); videoOutput.reset(); audioOutput.reset(); video.reset(); audio.reset(); - // Fresh codecs retain independent IDRs/AAC priming. One recorded network - // demuxer atomically rewinds both packet lanes, seeking upstream only on misses. + // Random access resets codec history; the network demuxer rewinds both + // packet lanes atomically, seeking upstream only on cache misses. const double audioTarget = audioIndex >= 0 && !remuxAudio(*audioSource->format->streams[audioIndex]->codecpar) ? aacPreroll(start) : NAN; if (videoInput->buffered) @@ -587,13 +630,15 @@ struct Pipeline { if (audioInput && audioInput->buffered) audioInput->seek(start, audioTarget); if (videoIndex >= 0) video = std::make_unique(*videoInput, videoIndex, AVMEDIA_TYPE_VIDEO, start, live, aligned); if (audioIndex >= 0) audio = std::make_unique(*audioSource, audioIndex, AVMEDIA_TYPE_AUDIO, start, live); + nextStart = start; } Segment segment(int64_t id, double start, double end) { - if (live) armOperation(); + armOperation(); Mux mux; if (video) { - videoOutput.reset(); - videoOutput = std::make_unique(*video, mux, false); + if (live) videoOutput.reset(); + if (videoOutput) videoOutput->beginSegment(mux); + else videoOutput = std::make_unique(*video, mux, live); } if (audio) { if (audioOutput) audioOutput->beginSegment(mux); @@ -604,6 +649,8 @@ struct Pipeline { av_dict_set(&options, "mpegts_copyts", "1", 0); av_dict_set(&options, "muxdelay", "0", 0); int result = avformat_write_header(mux.format, &options); av_dict_free(&options); check(result); + if (videoOutput) videoOutput->beginPackets(start, end); + if (audioOutput) audioOutput->beginPackets(start, end); const double audioEnd = audio && !audio->remux && !live ? double(aacBoundary(end)) / 48000 : end; for (;;) { require(!cancel && Clock::now() < deadline); @@ -613,12 +660,13 @@ struct Pipeline { if (!std::isfinite(vt) && !std::isfinite(at)) break; if (vt <= at) videoOutput->consume(start, end); else audioOutput->consume(start, end); } - if (videoOutput) videoOutput->finish(end); - if (audioOutput) audioOutput->finish(end); + if (videoOutput) videoOutput->finish(start, end); + if (audioOutput) audioOutput->finish(start, end); check(av_write_trailer(mux.format)); avio_flush(mux.io); require(mux.io->error >= 0 && !mux.bytes.empty()); if (videoInput->buffered) videoInput->buffered->updatePosition(end); if (audioInput && audioInput->buffered) audioInput->buffered->updatePosition(end); + nextStart = end; releaseOperation(); return {id, end - start, std::move(mux.bytes)}; } @@ -740,7 +788,7 @@ struct CastMedia { for (int64_t id = first; id < first + count; ++id) { char value[64]; double length = source.live ? segments[size_t(id - first)]->duration : boundary(id + 1) - boundary(id); std::snprintf(value, sizeof(value), "#EXTINF:%.6f,\n", length); - // Each independently generated segment resets codec/mux state. + // Each segment retains an independent mux and random-access entry. if (id > first) text += "#EXT-X-DISCONTINUITY\n"; text += value; text += "segment/" + std::to_string(id) + ".ts\n"; } @@ -761,8 +809,19 @@ struct CastMedia { if (auto segment = cached()) return segment; try { const double start = boundary(id), end = boundary(id + 1); - if (vodPipeline) vodPipeline->reset(start, !boundaries.empty()); - else vodPipeline = std::make_unique(source, cancelled, start); + if (vodPipeline) { + const auto &video = vodPipeline->video; + // Fixed-grid remux is safe only when the next boundary is an + // actual IDR. Otherwise reset once into the transcode path. + const bool independent = !video || !video->remux || + (video->pending && std::abs(video->packetTime() - start) < .001 && + (video->pending->flags & AV_PKT_FLAG_KEY) && + hasIDR(*video->pending, *video->stream->codecpar)); + const bool drained = (vodPipeline->videoOutput && vodPipeline->videoOutput->drained) || + (vodPipeline->audioOutput && vodPipeline->audioOutput->drained); + if (vodPipeline->nextStart != start || !independent || drained) + vodPipeline->reset(start, !boundaries.empty()); + } else vodPipeline = std::make_unique(source, cancelled, start); auto segment = std::make_shared(vodPipeline->segment(id, start, end)); { std::lock_guard guard(lock); segments.push_back(segment); while (segments.size() > windowSegments) segments.pop_front(); } return segment; diff --git a/TODO.md b/TODO.md index 0d156de..bd279ca 100644 --- a/TODO.md +++ b/TODO.md @@ -6,7 +6,8 @@ - [ ] evaluate playback optimization opportunities for YouTube and Miruro — measure benefits before changing behavior - [x] coalesce superseded, not-yet-started seeks in the player and Cast controls; preserve accumulated relative skips and play/pause ordering - [x] expand Cast H.264/AAC remux coverage with keyframe-aligned segments; preserve seek accuracy, timestamps and receiver compatibility - - [ ] reuse decoder/encoder state during sequential Cast playback; reset on random seeks and preserve independent keyframes and AAC continuity + - [x] reuse decoder/encoder state during sequential Cast playback; reset on random seeks and preserve independent keyframes and AAC continuity + - `.build/cast-media-check --codec-work`: 60s of 720p/30 B-frame video plus 44.1kHz PCM→AAC over loopback HTTP; three paired runs reduced median relay CPU from 3.56s to 2.58s (27%) and generation time from 4.97s to 4.43s (11%); physical-receiver playback still needs manual verification - [ ] evaluate VideoToolbox hardware video decoding for native and Cast playback; retain software fallback for unsupported formats - [ ] benchmark cold start, cached/uncached seeks and sustained playback separately to prioritize the actual bottlenecks - do not enable Miruro HTTP persistence without preserving origin/header checks, or increase caches/shorten Cast segments without measuring the memory/request/codec tradeoffs