diff --git a/Checks/cast-media.cpp b/Checks/cast-media.cpp index 4db5180..50102c3 100644 --- a/Checks/cast-media.cpp +++ b/Checks/cast-media.cpp @@ -10,13 +10,17 @@ extern "C" { #include #include #include +#include +#include #include #include #include #include +#include #include #include #include +#include #include #include #include @@ -39,12 +43,12 @@ static void writeFile(const std::string &path, const std::vector &bytes std::ofstream file(path, std::ios::binary); file.write(reinterpret_cast(bytes.data()), bytes.size()); assert(file.good()); } // Real source A/V built with the same bundled encoders; no PATH ffmpeg or fixtures. -static void fixture(const std::string &path) { +static void fixture(const std::string &path, int seconds = 90, int width = 96, int height = 64, int bitrate = 300000) { AVFormatContext *format = nullptr; ok(avformat_alloc_output_context2(&format, nullptr, "mpegts", path.c_str())); AVCodecContext *video = avcodec_alloc_context3(avcodec_find_encoder_by_name("h264_videotoolbox")); AVCodecContext *audio = avcodec_alloc_context3(avcodec_find_encoder(AV_CODEC_ID_AAC)); assert(video && audio); - video->width = 96; video->height = 64; video->pix_fmt = AV_PIX_FMT_NV12; - video->time_base = {1, 10}; video->framerate = {10, 1}; video->bit_rate = 300000; video->gop_size = 70; video->max_b_frames = 0; + video->width = width; video->height = height; video->pix_fmt = AV_PIX_FMT_NV12; + video->time_base = {1, 10}; video->framerate = {10, 1}; video->bit_rate = bitrate; video->gop_size = 70; video->max_b_frames = 0; video->sample_aspect_ratio = {4, 3}; av_opt_set(video->priv_data, "allow_sw", "1", 0); av_opt_set(video->priv_data, "realtime", "1", 0); audio->sample_rate = 48000; audio->sample_fmt = AV_SAMPLE_FMT_FLTP; audio->time_base = {1, 48000}; audio->bit_rate = 128000; @@ -80,14 +84,14 @@ static void fixture(const std::string &path) { av_packet_free(&packet); }; int64_t vi = 0, ai = 0; - while (vi < 900 || ai < 90 * 48000) { - bool isVideo = vi < 900 && (ai >= 90 * 48000 || vi / 10.0 <= ai / 48000.0); + while (vi < seconds * 10 || ai < seconds * 48000) { + bool isVideo = vi < seconds * 10 && (ai >= seconds * 48000 || vi / 10.0 <= ai / 48000.0); auto *f = av_frame_alloc(); assert(f); if (isVideo) { - f->width = 96; f->height = 64; f->format = video->pix_fmt; f->pts = vi++; + f->width = width; f->height = height; f->format = video->pix_fmt; f->pts = vi++; ok(av_frame_get_buffer(f, 32)); - for (int y = 0; y < 64; ++y) std::memset(f->data[0] + y * f->linesize[0], 40 + vi % 160, 96); - for (int y = 0; y < 32; ++y) std::memset(f->data[1] + y * f->linesize[1], 128, 96); + for (int y = 0; y < height; ++y) std::memset(f->data[0] + y * f->linesize[0], 40 + vi % 160, width); + for (int y = 0; y < height / 2; ++y) std::memset(f->data[1] + y * f->linesize[1], 128, width); encode(video, vs, f); } else { f->sample_rate = 48000; f->format = audio->sample_fmt; f->nb_samples = audio->frame_size; @@ -117,6 +121,27 @@ static void splitFixture(const std::string &source, const std::string &path, AVM int status = 0; assert(waitpid(child, &status, 0) == child); assert(WIFEXITED(status) && WEXITSTATUS(status) == 0); } +static std::map> hlsFixture(const std::string &source, const std::string &unusedSource, const std::filesystem::path &directory) { + std::filesystem::create_directory(directory); + for (const auto &variant : {std::pair{"selected", source}, std::pair{"unused", unusedSource}}) { + const auto playlist = (directory / (std::string(variant.first) + ".m3u8")).string(); + const auto segments = (directory / (std::string(variant.first) + "-%03d.ts")).string(); + const char *arguments[] = {"ffmpeg", "-hide_banner", "-loglevel", "error", "-y", + "-i", variant.second.c_str(), "-map", "0", "-c", "copy", "-hls_time", "6", + "-hls_list_size", "0", "-hls_segment_filename", segments.c_str(), playlist.c_str(), nullptr}; + pid_t child = 0; + assert(posix_spawnp(&child, arguments[0], nullptr, nullptr, const_cast(arguments), environ) == 0); + int status = 0; assert(waitpid(child, &status, 0) == child); + assert(WIFEXITED(status) && WEXITSTATUS(status) == 0); + } + std::map> resources; + for (const auto &entry : std::filesystem::directory_iterator(directory)) + resources["/hls/" + entry.path().filename().string()] = readFile(entry.path().string()); + const std::string master = "#EXTM3U\n#EXT-X-STREAM-INF:BANDWIDTH=400000,RESOLUTION=96x64,CODECS=\"avc1.64000a,mp4a.40.2\"\nselected.m3u8\n" + "#EXT-X-STREAM-INF:BANDWIDTH=200000,RESOLUTION=48x32,CODECS=\"avc1.64000a,mp4a.40.2\"\nunused.m3u8\n"; + resources["/hls/master.m3u8"] = {master.begin(), master.end()}; + return resources; +} static std::vector wav() { std::vector bytes(44 + 90 * 48000 * 2); auto put = [&](size_t at, uint32_t value, int length) { for (int i = 0; i < length; ++i) bytes[at + i] = uint8_t(value >> (8 * i)); }; @@ -129,7 +154,7 @@ static std::vector wav() { struct HTTP { int status = 0; std::string headers; std::vector bytes; }; -static HTTP request(const std::string &url, const std::string &method = "GET", const std::string &headers = "") { +static HTTP request(const std::string &url, const std::string &method = "GET", const std::string &headers = "", bool allowClosed = false, std::promise *sent = nullptr) { assert(url.starts_with("http://127.0.0.1:")); size_t slash = url.find('/', 7), colon = url.find(':', 7); sockaddr_in address{}; address.sin_family = AF_INET; address.sin_addr.s_addr = htonl(INADDR_LOOPBACK); address.sin_port = htons(std::stoi(url.substr(colon + 1, slash - colon - 1))); int socket = ::socket(AF_INET, SOCK_STREAM, 0); assert(socket >= 0); @@ -137,9 +162,17 @@ static HTTP request(const std::string &url, const std::string &method = "GET", c assert(connect(socket, reinterpret_cast(&address), sizeof(address)) == 0); std::string text = method + " " + url.substr(slash) + " HTTP/1.1\r\nHost: 127.0.0.1\r\n" + headers + "Connection: close\r\n\r\n"; assert(send(socket, text.data(), text.size(), 0) == ssize_t(text.size())); + if (sent) sent->set_value(); std::vector bytes; uint8_t buffer[65536]; - for (;;) { ssize_t n = recv(socket, buffer, sizeof(buffer), 0); assert(n >= 0); if (!n) break; bytes.insert(bytes.end(), buffer, buffer + n); } - close(socket); std::string raw(bytes.begin(), bytes.end()); size_t end = raw.find("\r\n\r\n"); assert(end != std::string::npos); + for (;;) { + ssize_t n = recv(socket, buffer, sizeof(buffer), 0); + if (n < 0 && allowClosed) break; + assert(n >= 0); if (!n) break; + bytes.insert(bytes.end(), buffer, buffer + n); + } + close(socket); std::string raw(bytes.begin(), bytes.end()); size_t end = raw.find("\r\n\r\n"); + if (allowClosed && end == std::string::npos) return {}; + assert(end != std::string::npos); return {std::stoi(raw.substr(9, 3)), raw.substr(0, end + 4), {bytes.begin() + end + 4, bytes.end()}}; } // A tiny real upstream tests resource-specific private headers and byte-range seek. @@ -148,6 +181,15 @@ struct Upstream { std::atomic foreignRequests{0}, videoRequests{0}, audioRequests{0}; std::atomic videoStarts{0}, audioStarts{0}; std::map> resources; + std::mutex mutex; + std::condition_variable gate; + std::map mediaRequests; + bool gated = false; + int held = 0, released = 0; + std::map counts() { + std::lock_guard lock(mutex); + return mediaRequests; + } std::string foreign; explicit Upstream(std::vector video, std::vector audio) { resources["/source.ts"] = std::move(video); resources["/audio.wav"] = std::move(audio); @@ -182,7 +224,19 @@ struct Upstream { bool authorized = publicMedia || request.find(audio ? "X-Audio: audio-secret" : "X-Video: video-secret") != std::string::npos; bool wrong = request.find(publicMedia ? "secret" : (audio ? "video-secret" : "audio-secret")) != std::string::npos; if (!authorized || wrong) { ++foreignRequests; status = 403; } - else { if (audio) ++audioRequests; else ++videoRequests; data = resource->second; } + else { + if (audio) ++audioRequests; else ++videoRequests; + data = resource->second; + if (path.starts_with("/hls/") && path.ends_with(".ts") && request.starts_with("GET")) { + std::unique_lock lock(mutex); + ++mediaRequests[path]; + if (gated) { + int ticket = ++held; + gate.notify_all(); + gate.wait(lock, [&] { return released >= ticket || stopped; }); + } + } + } } else status = 404; size_t low = 0, high = data.size(); auto range = request.find("Range: bytes="); @@ -201,14 +255,38 @@ struct Upstream { close(client); } }); } - ~Upstream() { stopped = true; shutdown(listener, SHUT_RDWR); for (auto &thread : threads) if (thread.joinable()) thread.join(); close(listener); } + ~Upstream() { + { std::lock_guard lock(mutex); stopped = true; gate.notify_all(); } + shutdown(listener, SHUT_RDWR); + for (auto &thread : threads) if (thread.joinable()) thread.join(); + close(listener); + } }; -static void playable(const HTTP &response, const std::string &path, bool video, double minimumTime) { +// Every media GET really waits at the upstream until this driver releases it. +// Deadlines are watchdogs, not media clocks; no elapsed-time sleeps drive the test. +template static auto throughGate(Upstream &upstream, Operation operation) { + bool done = false; + auto task = std::async(std::launch::async, [&] { + auto result = operation(); + { std::lock_guard lock(upstream.mutex); done = true; upstream.gate.notify_all(); } + return result; + }); + std::unique_lock lock(upstream.mutex); + while (!done) { + assert(upstream.gate.wait_for(lock, std::chrono::seconds(60), [&] { return done || upstream.held > upstream.released; })); + upstream.released = upstream.held; + upstream.gate.notify_all(); + } + lock.unlock(); + return task.get(); +} +static void playable(const HTTP &response, const std::string &path, bool video, double minimumTime, double maximumTime = INFINITY) { assert(response.status == 200 && response.bytes.size() > 188 && response.bytes[0] == 0x47); writeFile(path, response.bytes); AVFormatContext *format = nullptr; ok(avformat_open_input(&format, path.c_str(), nullptr, nullptr)); ok(avformat_find_stream_info(format, nullptr)); int audioIndex = av_find_best_stream(format, AVMEDIA_TYPE_AUDIO, -1, -1, nullptr, 0); assert(audioIndex >= 0); int videoIndex = av_find_best_stream(format, AVMEDIA_TYPE_VIDEO, -1, -1, nullptr, 0); assert(!video || videoIndex >= 0); std::map codecs; std::map frames; + std::map first; for (int index : {audioIndex, videoIndex}) if (index >= 0) { auto *stream = format->streams[index]; auto *codec = avcodec_alloc_context3(avcodec_find_decoder(stream->codecpar->codec_id)); assert(codec); ok(avcodec_parameters_to_context(codec, stream->codecpar)); ok(avcodec_open2(codec, codec->codec, nullptr)); codecs[index] = codec; @@ -217,10 +295,16 @@ static void playable(const HTTP &response, const std::string &path, bool video, while (av_read_frame(format, packet) >= 0) { auto found = codecs.find(packet->stream_index); if (found != codecs.end()) { - if (packet->pts != AV_NOPTS_VALUE) { double time = packet->pts * av_q2d(format->streams[packet->stream_index]->time_base); assert(time >= minimumTime - .1); timestamp = true; } + if (packet->pts != AV_NOPTS_VALUE) { + double time = packet->pts * av_q2d(format->streams[packet->stream_index]->time_base); + assert(time >= minimumTime - .1 && time <= maximumTime + .1); + if (!first.contains(packet->stream_index)) first[packet->stream_index] = time; + timestamp = true; + } ok(avcodec_send_packet(found->second, packet)); while (avcodec_receive_frame(found->second, frame) >= 0) { if (packet->stream_index == videoIndex) { + if (std::isfinite(maximumTime)) assert(frame->height == 64); auto sar = av_guess_sample_aspect_ratio(format, format->streams[videoIndex], frame); if (sar.num <= 0 || sar.den <= 0) sar = {1, 1}; assert(sar.num > 0 && std::abs(double(frame->width) / frame->height * av_q2d(sar) - 2.0) < .01); @@ -231,6 +315,8 @@ static void playable(const HTTP &response, const std::string &path, bool video, av_packet_unref(packet); } assert(timestamp && frames[audioIndex] > 10 && (!video || frames[videoIndex] > 10)); + if (std::isfinite(maximumTime)) for (int index : {audioIndex, videoIndex}) if (index >= 0) + assert(first.contains(index) && std::abs(first[index] - minimumTime) < .1); for (auto &[index, codec] : codecs) avcodec_free_context(&codec); av_packet_free(&packet); av_frame_free(&frame); avformat_close_input(&format); } @@ -302,11 +388,108 @@ static void seekWork(Upstream &upstream, const std::string &segment) { assert(seeksDoNotReprobe); assert(repeatedHasNoIO && repeatedIdentical && replayHasNoIO && replayIdentical); } +static int mediaCount(const std::map &counts, const std::string &prefix) { + int total = 0; + for (const auto &[path, count] : counts) if (path.starts_with(prefix)) total += count; + return total; +} +static void retainedMedia(const char *stage, const std::map &before, const std::map &after) { + for (const auto &[path, count] : after) { + int previous = before.contains(path) ? before.at(path) : 0; + int allowed = path.starts_with("/hls/unused-") ? previous : std::max(previous, 1); + if (count > allowed) + std::cerr << "hls-work retention failure: stage=" << stage << " media=" << path + << " previous=" << previous << " actual=" << count << " allowed=" << allowed << std::endl; + if (path.starts_with("/hls/unused-")) assert(count == previous); + else assert(count <= std::max(previous, 1)); + } +} +static void hlsWork(Upstream &upstream, const std::string &segment) { + std::string hls = upstream.url("/hls/master.m3u8"), url; + double duration = 0; + CastMediaSource source{hls.c_str(), "X-Video: video-secret\r\n", nullptr, nullptr, false, true, false, 180}; + auto *media = throughGate(upstream, [&] { return start(source, url, duration); }); + assert(duration >= 179 && duration <= 182); + const auto prepared = upstream.counts(); + assert(mediaCount(prepared, "/hls/selected-") > 0 && upstream.foreignRequests == 0); + const std::string base = url.substr(0, url.rfind('/') + 1); + auto fetch = [&](int id) { + const auto before = upstream.counts(); + std::cout << "hls-work requesting segment=" << id << " position=" << id * 6 << std::endl; + auto response = throughGate(upstream, [&] { return request(base + "segment/" + std::to_string(id) + ".ts"); }); + for (const auto &[path, count] : upstream.counts()) { + int previous = before.contains(path) ? before.at(path) : 0; + if (count != previous) + std::cout << "hls-work segment=" << id << " media=" << path + << " previous=" << previous << " actual=" << count << std::endl; + } + // The relay's shared positive MPEG-TS clock origin is one second; media positions remain id * 6. + playable(response, segment, true, id * 6 + 1, (id + 1) * 6 + 1); + return response; + }; + auto first = fetch(2), adjacent = fetch(3); + auto firstAudio = audioSpan(first, segment), adjacentAudio = audioSpan(adjacent, segment); + assert(std::abs(adjacentAudio.first - firstAudio.second) < 1.0 / 48000); + fetch(1); // New output segment, but both A/V lanes must rewind retained source packets. + assert(fetch(2).bytes == first.bytes); + const auto early = upstream.counts(); + std::cout << "hls-work media GETs after preparation: selected=" + << mediaCount(early, "/hls/selected-") - mediaCount(prepared, "/hls/selected-") + << " unused=" << mediaCount(early, "/hls/unused-") - mediaCount(prepared, "/hls/unused-") << std::endl; + retainedMedia("prepared -> early 2,3,1,2", prepared, early); + + // Beyond the 60-second read-ahead window: this random seek requires real upstream IO. + auto distant = fetch(24); + const auto distantCounts = upstream.counts(); + assert(mediaCount(distantCounts, "/hls/selected-") > mediaCount(early, "/hls/selected-")); + retainedMedia("early -> random 24", early, distantCounts); + auto distantAdjacent = fetch(26); + auto nearby = fetch(25); // Unrendered nearby output rewinds already-read source, not speculative history before a miss. + auto nearbyAudio = audioSpan(nearby, segment), distantAdjacentAudio = audioSpan(distantAdjacent, segment); + assert(std::abs(distantAdjacentAudio.first - nearbyAudio.second) < 1.0 / 48000); + assert(fetch(24).bytes == distant.bytes); + const auto nearbyCounts = upstream.counts(); + retainedMedia("random 24 -> 26,25,24", distantCounts, nearbyCounts); + + // Behind the 30-second retained window: a backward random miss must still decode at the target. + auto backward = fetch(5); + const auto backwardCounts = upstream.counts(); + assert(mediaCount(backwardCounts, "/hls/selected-") > mediaCount(nearbyCounts, "/hls/selected-")); + assert(mediaCount(backwardCounts, "/hls/unused-") == mediaCount(prepared, "/hls/unused-")); + assert(fetch(5).bytes == backward.bytes); + retainedMedia("backward 5 -> replay 5", backwardCounts, upstream.counts()); + + // Do not release the next media request. Cancel/destroy must finish while upstream is held. + std::promise sent; + auto sentRequest = sent.get_future(); + auto blocked = std::async(std::launch::async, [&] { return request(base + "segment/20.ts", "GET", "", true, &sent); }); + assert(sentRequest.wait_for(std::chrono::seconds(10)) == std::future_status::ready); + int released = 0; + { + std::unique_lock lock(upstream.mutex); + assert(upstream.gate.wait_for(lock, std::chrono::seconds(10), [&] { return upstream.held > upstream.released; })); + released = upstream.released; + } + assert(blocked.wait_for(std::chrono::seconds(0)) == std::future_status::timeout); + auto destroyed = std::async(std::launch::async, [&] { cast_media_cancel(media); cast_media_destroy(media); }); + assert(destroyed.wait_for(std::chrono::seconds(3)) == std::future_status::ready); + destroyed.get(); + assert(blocked.wait_for(std::chrono::seconds(3)) == std::future_status::ready); + auto cancelled = blocked.get(); assert(cancelled.status == 0 || cancelled.status == 503); + { + std::lock_guard lock(upstream.mutex); + assert(upstream.released == released && upstream.held > upstream.released); + upstream.released = upstream.held; + upstream.gate.notify_all(); + } + assert(upstream.foreignRequests == 0); +} int main(int argc, char **argv) { - bool liveCheck = false, seekCheck = false; + bool liveCheck = false, seekCheck = false, hlsCheck = false; for (int argument = 1; argument < argc; ++argument) { if (std::string(argv[argument]) == "--live") liveCheck = true; if (std::string(argv[argument]) == "--seek-work") seekCheck = true; + if (std::string(argv[argument]) == "--hls-work") hlsCheck = true; } char temp[] = "/tmp/terebi-cast-check.XXXXXX"; assert(mkdtemp(temp)); std::filesystem::path directory(temp); const auto video = (directory / "source.ts").string(), audio = (directory / "source.wav").string(), segment = (directory / "segment.ts").string(); @@ -325,6 +508,18 @@ int main(int argc, char **argv) { std::cout << "cast-media: split VOD seek work passed\n"; return 0; } + const auto selectedHLS = (directory / "selected.ts").string(), unusedHLS = (directory / "unused.ts").string(); + fixture(selectedHLS, 180); + fixture(unusedHLS, 180, 48, 32, 100000); + Upstream delayed(videoBytes, audioBytes); + delayed.resources.merge(hlsFixture(selectedHLS, unusedHLS, directory / "hls")); + delayed.gated = true; delayed.start(); + hlsWork(delayed, segment); + if (hlsCheck) { + std::filesystem::remove_all(directory); + std::cout << "cast-media: authenticated delayed multi-variant combined HLS passed\n"; + return 0; + } std::string url; double duration = 0; CastMediaSource local{video.c_str(), "", nullptr, nullptr, true, false, false, 0}; auto *media = start(local, url, duration); assert(duration >= 89 && duration <= 92); @@ -393,5 +588,5 @@ int main(int argc, char **argv) { playable(request(base + "segment/" + std::to_string(last) + ".ts"), segment, true, 0); cast_media_destroy(media); } std::filesystem::remove_all(directory); - std::cout << "cast-media: playable A/V, VOD random seek, separate audio, HLS origin/header policy, subtitles, ranges, capabilities, cancellation" << (liveCheck ? ", bounded live window" : " (use --live for rolling window)") << " passed\n"; + std::cout << "cast-media: playable A/V, VOD random seek, separate audio, authenticated delayed multi-variant HLS retention, HLS origin/header policy, subtitles, ranges, capabilities, gated cancellation" << (liveCheck ? ", bounded live window" : " (use --live for rolling window)") << " passed\n"; } diff --git a/Makefile b/Makefile index 0d7691e..2addf62 100644 --- a/Makefile +++ b/Makefile @@ -68,7 +68,7 @@ $(BUILD_DIR)/player.o: Sources/Core/player.cpp Sources/Core/player.h Sources/Cor $(BUILD_DIR)/progress_reporter.o: Sources/Core/progress_reporter.cpp Sources/Core/progress_reporter.hpp Sources/Core/player.h Makefile $(NATIVE_STAMP) | $(BUILD_DIR) $(CXX) $(CXXFLAGS) -c $< -o $@ -$(BUILD_DIR)/cast_media.o: Sources/Core/cast_media.cpp Sources/Core/cast_media.h Sources/Core/network_policy.hpp Makefile $(NATIVE_STAMP) | $(BUILD_DIR) check-dependencies +$(BUILD_DIR)/cast_media.o: Sources/Core/cast_media.cpp Sources/Core/cast_media.h Sources/Core/network_policy.hpp Sources/Core/network_input.hpp Makefile $(NATIVE_STAMP) | $(BUILD_DIR) check-dependencies $(CXX) $(CXXFLAGS) -c $< -o $@ $(BUILD_DIR)/Output.o: Sources/macOS/Output.mm Sources/Core/output.hpp Makefile $(NATIVE_STAMP) | $(BUILD_DIR) check-dependencies @@ -230,7 +230,7 @@ check-media: $(BUILD_DIR)/system-playback-check check-fixtures $(BUILD_DIR)/cast-session-check: Checks/cast-session.swift Sources/App/CastSession.swift Makefile | $(BUILD_DIR) $(SWIFT) -target $(ARCH)-apple-macosx15.0 -swift-version 6 -parse-as-library -DCAST_SESSION_CHECKS -O $(DEBUG_FLAGS) Checks/cast-session.swift Sources/App/CastSession.swift -framework Foundation -framework Network -framework Security -o $@ -$(BUILD_DIR)/cast_media-check.o: Sources/Core/cast_media.cpp Sources/Core/cast_media.h Sources/Core/network_policy.hpp Makefile $(NATIVE_STAMP) | $(BUILD_DIR) check-dependencies +$(BUILD_DIR)/cast_media-check.o: Sources/Core/cast_media.cpp Sources/Core/cast_media.h Sources/Core/network_policy.hpp Sources/Core/network_input.hpp Makefile $(NATIVE_STAMP) | $(BUILD_DIR) check-dependencies $(CXX) $(CXXFLAGS) -DCAST_MEDIA_CHECK -c $< -o $@ $(BUILD_DIR)/cast-media-check: Checks/cast-media.cpp $(BUILD_DIR)/cast_media-check.o Makefile | check-dependencies diff --git a/README.md b/README.md index 9ce11a8..4165542 100644 --- a/README.md +++ b/README.md @@ -217,7 +217,7 @@ The Mac serves session-only capability URLs over HTTP. Keep it awake and connect Local files, resolved Miruro HLS and YouTube's combined/separate video/audio inputs are delivered as six-second HLS segments. Compatible, independently decodable H.264/AAC is remuxed; other supported inputs use VideoToolbox H.264 (including its platform software fallback) and stereo AAC. Output targets at most **1920×1080 at 30 fps**, H.264 Main/High level 4.1 or lower, with preserved display aspect ratio. Recorded segments are generated on demand, including distant seeks; there is no full-movie download or preprocessing step. Live output keeps eight segments (about 48 seconds) and is not seekable. Audio-only sources are supported. -Recorded delivery retains already-probed upstream inputs and caches eight completed segments (at most 256 MiB). Repeated, HEAD and range requests reuse cached bytes; uncached segments seek the retained inputs rather than reopening and probing both streams. Uncached generation is serialized to keep FFmpeg input ownership simple. GOP preroll, transcoding and receiver buffering can still make a fresh Cast seek slower than local playback. +Recorded network delivery uses the native player's bounded compressed-packet buffer: up to 128 MiB shared between inputs, targeting sixty seconds ahead and thirty seconds behind (byte pressure can shorten retention). Combined HLS supplies video/audio from one selected-stream reader; unused quality variants are disabled. Sequential and nearby seeks reuse packets, including decoder/AAC preroll; true cache misses seek upstream once. Nearby forward requests wait for existing read-ahead instead of cancelling and restarting its fetch. Eight completed output segments are also cached (at most 256 MiB), so repeated, HEAD and range requests reuse encoded bytes. Uncached generation is serialized; GOP preroll, transcoding, provider slowness and receiver buffering can still delay a fresh Cast seek. Miruro retains its HTTP(S)-only, same-origin media and no-redirect policy. Public YouTube HLS can cross CDN origins, but supplied headers never follow it to another origin. All Cast media redirects are rejected. WebVTT is fetched with each track's own headers and the existing bounded, same-origin subtitle redirect policy, then served with CORS. Private upstream addresses and headers are not sent to the receiver or included in errors. @@ -225,6 +225,8 @@ The Cast channel accepts the selected receiver's self-signed TLS certificate: en `make check-cast` exercises a real self-signed TLS loopback receiver, decoded relay output for all three source paths, header/origin isolation, recorded seeking, VOD/live audio continuity, anamorphic video, capability expiry, controlled protocol deadlines/live pacing, and complete player handoff/back-to-Mac recovery. Regression checks cover slow/asynchronous launch, stale volume feedback, native slider tracking and split-stream seek work: repeated segments add no upstream requests, and indexed distant/adjacent seeks add no origin-zero reprobes. A production-paced live HLS smoke also decoded A/V across a segment boundary and shut down. **YouTube Cast playback has been observed on a physical receiver; these fixes, receiver captions and perceived A/V synchronization still need human verification.** +Authenticated multi-variant HLS regressions gate upstream responses and verify no unused-quality downloads or duplicate retained-media reads, nearby rewinds, distant cache misses, decoded A/V positions and AAC continuity, and cancellation while upstream remains blocked. Native network playback checks cover the shared reader's existing callers. + ## Resume playback Reopening a local video, YouTube video or resolved Miruro episode automatically resumes its saved position, including after quitting and relaunching the app. Local files use their resolved file paths; YouTube uses the canonical video webpage URL; Miruro uses its opaque media ID and episode number. Newly resolved addresses therefore share progress without saving CDN tokens as identity. The app does not automatically reopen the last video on launch. diff --git a/Sources/Core/cast_media.cpp b/Sources/Core/cast_media.cpp index d1a6b66..d257c9e 100644 --- a/Sources/Core/cast_media.cpp +++ b/Sources/Core/cast_media.cpp @@ -1,5 +1,6 @@ #include "cast_media.h" #include "network_policy.hpp" +#include "network_input.hpp" extern "C" { #include #include @@ -84,8 +85,13 @@ struct Input { std::atomic &cancel; std::string origin, headers; std::atomic deadline{0}; - int64_t operationDeadline = std::numeric_limits::max(); + std::atomic operationDeadline{std::numeric_limits::max()}; bool network, restricted, live; + std::atomic epoch{0}; + std::mutex waitLock; + std::condition_variable wake; + bool positioned = false; + std::unique_ptr buffered; Input(std::atomic &stop, const std::string &url, const std::string &fields, bool local, bool restricted, bool live) : cancel(stop), headers(fields), network(!local), restricted(restricted), live(live) { origin = local ? "" : playback::networkOrigin(url.c_str()); @@ -115,14 +121,21 @@ struct Input { if (result >= 0) { arm(); result = avformat_find_stream_info(format, nullptr); } if (result < 0) { avformat_close_input(&format); check(result); } } - ~Input() { avformat_close_input(&format); } + ~Input() { + if (buffered) buffered->stop(); + else avformat_close_input(&format); + } void arm() { const auto ioDeadline = std::chrono::duration_cast((Clock::now() + std::chrono::seconds(15)).time_since_epoch()).count(); - deadline = std::min(ioDeadline, operationDeadline); + deadline = std::min(ioDeadline, operationDeadline.load()); + } + bool interrupted() const { + const auto now = std::chrono::duration_cast(Clock::now().time_since_epoch()).count(); + return cancel || now >= operationDeadline.load() || (!buffered && now >= deadline.load()); } static int interrupt(void *opaque) { auto &in = *static_cast(opaque); - return in.cancel || std::chrono::duration_cast(Clock::now().time_since_epoch()).count() > in.deadline; + return in.interrupted() || (in.buffered && playback::NetworkInput::interrupt(in.buffered.get())); } static int openResource(AVFormatContext *context, AVIOContext **io, const char *url, int flags, AVDictionary **options) noexcept { try { @@ -141,10 +154,48 @@ struct Input { } double offset() const { return format->start_time == AV_NOPTS_VALUE ? 0 : double(format->start_time) / AV_TIME_BASE; } double duration() const { return format->duration == AV_NOPTS_VALUE ? 0 : double(format->duration) / AV_TIME_BASE; } - void seek(double seconds) { - arm(); check(av_seek_frame(format, -1, int64_t((seconds + offset()) * AV_TIME_BASE), AVSEEK_FLAG_BACKWARD)); + void select(std::array indices, playback::StreamCacheBudget &budget) { + for (unsigned i = 0; i < format->nb_streams; ++i) + format->streams[i]->discard = int(i) == indices[0] || int(i) == indices[1] ? + AVDISCARD_DEFAULT : AVDISCARD_ALL; + if (!network || live) return; + buffered = std::make_unique(cancel, epoch, 0, wake, budget); + buffered->format = format; + buffered->start(indices, offset(), true); + } + void seek(double seconds, double audioTarget = NAN) { + const auto timestamp = int64_t((seconds + offset()) * AV_TIME_BASE); + if (buffered) { + // Probing retains the opening packets. Do not reopen their HLS media + // resource just to establish the initial zero-position cursors. + if (!positioned && seconds == 0) { positioned = true; return; } + positioned = true; + // A nearby forward request can outrun the asynchronous fetch frontier. + // Let the existing read complete rather than interrupting and refetching it. + while (buffered->readAheadPending(seconds, audioTarget)) { + require(!interrupted()); + std::unique_lock lock(waitLock); + wake.wait_for(lock, std::chrono::milliseconds(20)); + } + buffered->seek(-1, timestamp, audioTarget); + } else { arm(); check(av_seek_frame(format, -1, timestamp, AVSEEK_FLAG_BACKWARD)); } + } + int read(int slot, AVPacket *destination) { + if (!buffered) return av_read_frame(format, destination); + for (;;) { + require(!interrupted()); + double end = 0; + int result = buffered->take(slot, destination, end); + if (result != 0) return result == 1 ? 0 : result; + std::unique_lock lock(waitLock); + wake.wait_for(lock, std::chrono::milliseconds(20)); + } } }; +bool remuxAudio(const AVCodecParameters &codec) { + return codec.codec_id == AV_CODEC_ID_AAC && codec.ch_layout.nb_channels <= 2 && + (codec.profile == AV_PROFILE_AAC_LOW || codec.profile == AV_PROFILE_AAC_HE || codec.profile == AV_PROFILE_AAC_HE_V2); +} struct Track { Input *input; AVStream *stream = nullptr; @@ -153,9 +204,8 @@ struct Track { Frame decoded; bool eof = false, draining = false, remux = false; double nextTime = 0; - Track(Input &in, AVMediaType type, double start, bool live) : input(&in) { - int index = av_find_best_stream(input->format, type, -1, -1, nullptr, 0); - if (index < 0) return; + Track(Input &in, int index, AVMediaType type, double start, bool live) : input(&in) { + require(index >= 0); stream = input->format->streams[index]; remux = stream->codecpar->codec_id == (type == AVMEDIA_TYPE_VIDEO ? AV_CODEC_ID_H264 : AV_CODEC_ID_AAC); if (type == AVMEDIA_TYPE_VIDEO) { @@ -166,13 +216,12 @@ struct Track { stream->codecpar->width <= 1920 && stream->codecpar->height <= 1080 && rate.num > 0 && av_q2d(rate) <= 30.001; } else { - remux = remux && stream->codecpar->ch_layout.nb_channels <= 2 && - (stream->codecpar->profile == AV_PROFILE_AAC_LOW || stream->codecpar->profile == AV_PROFILE_AAC_HE || stream->codecpar->profile == AV_PROFILE_AAC_HE_V2); + remux = remuxAudio(*stream->codecpar); } const double seekTarget = type == AVMEDIA_TYPE_AUDIO && !remux ? std::max(0.0, aacPreroll(start) - double(std::max(0, stream->codecpar->seek_preroll)) / std::max(1, stream->codecpar->sample_rate)) : start; - if (!live) input->seek(seekTarget); + if (!live && !input->buffered) input->seek(seekTarget); nextPacket(); // Every VOD segment must start at its advertised time and on an IDR. If the // source GOP doesn't align, decode preroll and use VideoToolbox, not overlap. @@ -212,8 +261,8 @@ struct Track { auto p = packet(); input->arm(); for (;;) { - require(!Input::interrupt(input)); - int result = av_read_frame(input->format, p.get()); + require(!input->interrupted()); + int result = input->read(stream->codecpar->codec_type == AVMEDIA_TYPE_AUDIO ? 0 : 1, p.get()); if (result == AVERROR_EOF) { eof = true; return; } check(result); require(p->size >= 0 && size_t(p->size) <= segmentLimit); @@ -229,7 +278,7 @@ struct Track { void nextFrame() { decoded.reset(); auto f = frame(); for (;;) { - require(!Input::interrupt(input)); + require(!input->interrupted()); int result = avcodec_receive_frame(decoder, f.get()); if (result == 0) { decoded = std::move(f); @@ -440,44 +489,70 @@ struct OutputTrack { } }; struct Pipeline { + playback::StreamCacheBudget cacheBudget; std::unique_ptr videoInput, audioInput; + Input *audioSource = nullptr; + int videoIndex = -1, audioIndex = -1; + Clock::time_point deadline; std::unique_ptr video, audio; std::unique_ptr videoOutput, audioOutput; bool live; std::atomic &cancel; double duration = 0; Pipeline(const Source &source, std::atomic &stop, double start) : live(source.live), cancel(stop) { - auto primary = std::make_unique(stop, source.video, source.videoHeaders, source.local, source.restricted, source.live); - duration = std::max(source.duration, primary->duration()); - bool hasVideo = av_find_best_stream(primary->format, AVMEDIA_TYPE_VIDEO, -1, -1, nullptr, 0) >= 0; - bool hasAudio = av_find_best_stream(primary->format, AVMEDIA_TYPE_AUDIO, -1, -1, nullptr, 0) >= 0; - if (hasVideo) videoInput = std::move(primary); + videoInput = std::make_unique(stop, source.video, source.videoHeaders, source.local, source.restricted, source.live); + duration = std::max(source.duration, videoInput->duration()); + videoIndex = av_find_best_stream(videoInput->format, AVMEDIA_TYPE_VIDEO, -1, -1, nullptr, 0); if (!source.audio.empty()) { + require(videoIndex >= 0); audioInput = std::make_unique(stop, source.audio, source.audioHeaders, false, source.restricted, source.live); - } else if (hasAudio) { - if (!primary) primary = std::make_unique(stop, source.video, source.videoHeaders, source.local, source.restricted, source.live); - audioInput = std::move(primary); + audioSource = audioInput.get(); + audioIndex = av_find_best_stream(audioSource->format, AVMEDIA_TYPE_AUDIO, -1, -1, nullptr, 0); + } else { + audioIndex = av_find_best_stream(videoInput->format, AVMEDIA_TYPE_AUDIO, -1, videoIndex, nullptr, 0); + if (audioIndex >= 0) { + if (videoIndex >= 0 && (source.local || source.live)) { + audioInput = std::make_unique(stop, source.video, source.videoHeaders, source.local, source.restricted, source.live); + audioIndex = av_find_best_stream(audioInput->format, AVMEDIA_TYPE_AUDIO, -1, videoIndex, nullptr, 0); + audioSource = audioInput.get(); + } else audioSource = videoInput.get(); + } } + require(videoIndex >= 0 || audioIndex >= 0); + videoInput->select({audioSource == videoInput.get() ? audioIndex : -1, videoIndex}, cacheBudget); + if (audioInput) audioInput->select({audioIndex, -1}, cacheBudget); reset(start); if (!source.audio.empty()) require(audio && audio->stream); require(video || audio); if (!source.live) require(std::isfinite(duration) && duration > 0 && duration <= 86400 * 14); + releaseOperation(); + } + 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; + } + void releaseOperation() { + videoInput->operationDeadline = std::numeric_limits::max(); + if (audioInput) audioInput->operationDeadline = std::numeric_limits::max(); } void reset(double start) { + armOperation(); videoOutput.reset(); audioOutput.reset(); video.reset(); audio.reset(); - // Fresh codecs preserve independent segment/AAC priming; demuxers retain - // their already-probed metadata and seek indexes across random requests. - if (videoInput) { - videoInput->operationDeadline = std::numeric_limits::max(); - video = std::make_unique(*videoInput, AVMEDIA_TYPE_VIDEO, start, live); - } - if (audioInput) { - audioInput->operationDeadline = std::numeric_limits::max(); - audio = std::make_unique(*audioInput, AVMEDIA_TYPE_AUDIO, start, live); - } + // Fresh codecs retain independent IDRs/AAC priming. One recorded network + // demuxer atomically rewinds both packet lanes, seeking upstream only on misses. + const double audioTarget = audioIndex >= 0 && !remuxAudio(*audioSource->format->streams[audioIndex]->codecpar) ? + aacPreroll(start) : start; + if (videoInput->buffered) + videoInput->seek(start, audioSource == videoInput.get() ? audioTarget : NAN); + if (audioInput && audioInput->buffered) audioInput->seek(start, audioTarget); + if (videoIndex >= 0) video = std::make_unique(*videoInput, videoIndex, AVMEDIA_TYPE_VIDEO, start, live); + if (audioIndex >= 0) audio = std::make_unique(*audioSource, audioIndex, AVMEDIA_TYPE_AUDIO, start, live); } Segment segment(int64_t id, double start, double end) { + if (live) armOperation(); Mux mux; if (video) { videoOutput.reset(); @@ -492,10 +567,6 @@ 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); - const auto deadline = Clock::now() + std::chrono::seconds(45); - const auto limit = std::chrono::duration_cast(deadline.time_since_epoch()).count(); - if (video) video->input->operationDeadline = limit; - if (audio) audio->input->operationDeadline = limit; const double audioEnd = audio && !audio->remux && !live ? double(aacBoundary(end)) / 48000 : end; for (;;) { require(!cancel && Clock::now() < deadline); @@ -509,6 +580,9 @@ struct Pipeline { if (audioOutput) audioOutput->finish(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); + releaseOperation(); return {id, end - start, std::move(mux.bytes)}; } }; diff --git a/Sources/Core/network_input.hpp b/Sources/Core/network_input.hpp index 885e47c..0f739aa 100644 --- a/Sources/Core/network_input.hpp +++ b/Sources/Core/network_input.hpp @@ -125,7 +125,9 @@ public: for (int slot = 0; slot < 2; ++slot) trim(slot); budget.changed.notify_all(); } - void seek(int streamIndex, int64_t timestamp) { + // Optional audio target gives fresh segment encoders their AAC preroll without + // moving the video cursor away from its own keyframe. Omission preserves player seeks. + void seek(int streamIndex, int64_t timestamp, double audioTarget = NAN) { std::lock_guard lock(budget.mutex); const double target = (streamIndex < 0 ? static_cast(timestamp) / AV_TIME_BASE : static_cast(timestamp) * av_q2d(format->streams[streamIndex]->time_base)) - origin; @@ -138,7 +140,7 @@ public: // track can resume without restoring its I/O position through a real seek. if (halted && !ended && !(std::isfinite(lane.mediaEnd) && lane.frontierEnd + lane.tick >= lane.mediaEnd)) cached = false; - else cached = cachedCursor(slot, target, cursors[slot]); + else cached = cachedCursor(slot, slot == 0 && std::isfinite(audioTarget) ? audioTarget : target, cursors[slot]); } position = std::max(0.0, target); if (cached) { @@ -156,6 +158,11 @@ public: failure = 0; seekIndex = streamIndex; seekTimestamp = timestamp; + if (lanes[0].index >= 0 && !skipAudio && std::isfinite(audioTarget)) { + const double earliest = std::min(target, std::max(0.0, audioTarget - lanes[0].preroll)); + seekIndex = -1; + seekTimestamp = static_cast((earliest + origin) * AV_TIME_BASE); + } seeking = true; serial.fetch_add(1, std::memory_order_release); budget.changed.notify_all(); @@ -187,6 +194,20 @@ public: std::lock_guard lock(budget.mutex); return failure; } + // Whether an existing bounded forward fetch can supply a seek's missing packets. + // Past history and targets outside the current ahead window require normal seek handling. + bool readAheadPending(double target, double audioTarget = NAN) const { + std::lock_guard lock(budget.mutex); + if (failure || seeking || ended || halted || waitingForRoom || + target < position || target > position + aheadSeconds) return false; + for (int slot = 0; slot < 2; ++slot) { + const auto &lane = lanes[slot]; + if (lane.index < 0 || (slot == 0 && skipAudio) || trackEnded(lane)) continue; + const double laneTarget = slot == 0 && std::isfinite(audioTarget) ? audioTarget : target; + if (laneTarget > lane.frontierEnd + lane.tick) return true; + } + return false; + } double bufferedStart(int slot) const { std::lock_guard lock(budget.mutex); if (!cacheSeeking || failure || seeking || (slot == 0 && skipAudio)) return NAN; @@ -253,6 +274,7 @@ private: std::atomic deadline{std::numeric_limits::max()}; bool seeking = false, ended = false, halted = false, skipAudio = false; bool cacheSeeking = false, started = false, audioDropped = false; + bool waitingForRoom = false; int seekIndex = -1, failure = 0; int64_t seekTimestamp = 0; double origin = 0, position = 0; @@ -464,10 +486,15 @@ private: } // The demuxer's single in-flight packet is not retained until it fits. // Byte pressure shortens consumed history, never drops unread packets. - budget.changed.wait(lock, [&] { - return stoppingNow() || serial.load(std::memory_order_acquire) != current || - (slot == 0 && skipAudio) || makeRoom(lane, bytes); - }); + if (!makeRoom(lane, bytes)) { + waitingForRoom = true; + playerWake.notify_one(); + budget.changed.wait(lock, [&] { + return stoppingNow() || serial.load(std::memory_order_acquire) != current || + (slot == 0 && skipAudio) || makeRoom(lane, bytes); + }); + waitingForRoom = false; + } if (stoppingNow()) return; if (serial.load(std::memory_order_acquire) != current || (slot == 0 && skipAudio)) { if (slot == 0 && skipAudio) audioDropped = true; diff --git a/TODO.md b/TODO.md index 7df6c3c..f8be3d6 100644 --- a/TODO.md +++ b/TODO.md @@ -4,5 +4,6 @@ - [x] fix premature receiver-launch timeout: reports failure even though Chromecast subsequently connects and playback works - [x] fix TV volume slider jitter: receiver updates appear to compete with local volume adjustments - [x] investigate and reduce slow buffering after seeking in YouTube videos while casting, compared with local playback — reuse probed inputs and cache completed segments; confirm perceived improvement on TV + - [x] fix Miruro Cast upstream overfetch: select HLS streams and share bounded video/audio packet buffering — verify sustained playback on TV - [ ] in the "choose what to play" screen, we should change the youtube button general streaming - we can figure out the stream and how to handle it from the url (i.e. if youtube url, handle as youtube stream; if miruro url, handle accordingly) \ No newline at end of file