hl-valdemar/terebi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Releases Wiki Activity

fix: share buffered Cast HLS input and select streams

Browse Source
This commit is contained in:
Valdemar Høgsvig Lorenzen
2026-10-13 22:47:07 +02:00
parent b0636c5af7
commit ad4b2da5c8
6 changed files with 362 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Checks/cast-media.cpp
+212 -17
View File
@@ -10,13 +10,17 @@ extern "C" {
#include <cassert>
#include <cctype>
#include <cstdlib>
#include <chrono>
#include <condition_variable>
#include <cmath>
#include <cstring>
#include <filesystem>
#include <fstream>
#include <future>
#include <iostream>
#include <iterator>
#include <map>
#include <mutex>
#include <string>
#include <thread>
#include <vector>
@@ -39,12 +43,12 @@ static void writeFile(const std::string &path, const std::vector<uint8_t> &bytes
std::ofstream file(path, std::ios::binary); file.write(reinterpret_cast<const char *>(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<std::string, std::vector<uint8_t>> 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<char *const *>(arguments), environ) == 0);
int status = 0; assert(waitpid(child, &status, 0) == child);
assert(WIFEXITED(status) && WEXITSTATUS(status) == 0);
}
std::map<std::string, std::vector<uint8_t>> 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<uint8_t> wav() {
std::vector<uint8_t> 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<uint8_t> wav() {
struct HTTP {
int status = 0; std::string headers; std::vector<uint8_t> 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<void> *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<sockaddr *>(&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<uint8_t> 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<int> foreignRequests{0}, videoRequests{0}, audioRequests{0};
std::atomic<int> videoStarts{0}, audioStarts{0};
std::map<std::string, std::vector<uint8_t>> resources;
std::mutex mutex;
std::condition_variable gate;
std::map<std::string, int> mediaRequests;
bool gated = false;
int held = 0, released = 0;
std::map<std::string, int> counts() {
std::lock_guard lock(mutex);
return mediaRequests;
}
std::string foreign;
explicit Upstream(std::vector<uint8_t> video, std::vector<uint8_t> 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<typename Operation> 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<int, AVCodecContext *> codecs; std::map<int, int> frames;
std::map<int, double> 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<std::string, int> &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<std::string, int> &before, const std::map<std::string, int> &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<void> 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";
}