hl-valdemar/terebi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Releases Wiki Activity
Files
29a70678a45c8883c8b76fcd12b880ad66217950
terebi/Sources/Core/network_input.hpp
T

535 lines
24 KiB
C++
Raw Blame History

#pragma once
extern "C" {
#include <libavformat/avformat.h>
}
#include <array>
#include <algorithm>
#include <atomic>
#include <chrono>
#include <cmath>
#include <cerrno>
#include <condition_variable>
#include <cstdint>
#include <deque>
#include <iterator>
#include <limits>
#include <mutex>
#include <new>
#include <stdexcept>
#include <string>
#include <thread>
namespace playback {
class NetworkInput;
// One budget per player, shared by combined or separate compressed inputs.
class StreamCacheBudget {
public:
static constexpr std::size_t byteLimit = 128 * 1024 * 1024;
explicit StreamCacheBudget(std::size_t limit = byteLimit)
: limit(std::min(limit, byteLimit)) {
if (!this->limit) throw std::invalid_argument("The streaming cache budget must be positive");
}
std::size_t bytes() const {
std::lock_guard lock(mutex);
return used;
}
private:
friend class NetworkInput;
const std::size_t limit;
std::size_t used = 0;
mutable std::mutex mutex;
std::condition_variable changed;
std::array<NetworkInput *, 2> inputs{};
};
// Only network inputs need a reader: a blocked socket must never own the player worker.
// Format metadata is immutable after start(); the reader exclusively owns demux and seek I/O.
class NetworkInput {
public:
static constexpr double aheadSeconds = 60, behindSeconds = 30;
AVFormatContext *format = nullptr;
// Resolved sources bind credentials to their original origin for all HLS I/O.
std::string headerOrigin;
std::string requestHeaders;
NetworkInput(const std::atomic<bool> &stopping, const std::atomic<uint64_t> &epoch,
uint64_t activeEpoch, std::condition_variable &playerWake, StreamCacheBudget &budget)
: stopping(stopping), epoch(epoch), activeEpoch(activeEpoch), playerWake(playerWake), budget(budget) {
std::lock_guard lock(budget.mutex);
auto free = std::find(budget.inputs.begin(), budget.inputs.end(), nullptr);
if (free == budget.inputs.end()) throw std::invalid_argument("Too many streaming inputs");
*free = this;
}
~NetworkInput() {
stop();
{
std::lock_guard lock(budget.mutex);
for (auto &lane : lanes) clear(lane);
for (auto &input : budget.inputs) if (input == this) input = nullptr;
budget.changed.notify_all();
}
av_packet_free(&scratch);
avformat_close_input(&format);
}
NetworkInput(const NetworkInput &) = delete;
NetworkInput &operator=(const NetworkInput &) = delete;
static int interrupt(void *opaque) {
auto &input = *static_cast<NetworkInput *>(opaque);
return input.cancelled.load(std::memory_order_relaxed) ||
input.stopping.load(std::memory_order_relaxed) ||
input.epoch.load(std::memory_order_relaxed) != input.activeEpoch ||
input.serial.load(std::memory_order_acquire) != input.operationSerial.load(std::memory_order_relaxed) ||
now() >= input.deadline.load(std::memory_order_relaxed);
}
void armIO() {
operationSerial.store(serial.load(std::memory_order_acquire), std::memory_order_relaxed);
deadline.store(now() + 15'000'000'000LL, std::memory_order_relaxed);
}
void start(std::array<int, 2> indices, double timelineOrigin, bool allowCachedSeeking) {
{
std::lock_guard lock(budget.mutex);
origin = timelineOrigin;
cacheSeeking = allowCachedSeeking;
started = true;
for (int slot = 0; slot < 2; ++slot) {
auto &lane = lanes[slot];
lane.index = indices[slot];
if (lane.index < 0) continue;
const auto *stream = format->streams[lane.index];
lane.tick = av_q2d(stream->time_base);
lane.mediaStart = stream->start_time == AV_NOPTS_VALUE ? 0 :
stream->start_time * lane.tick - origin;
if (stream->duration != AV_NOPTS_VALUE && stream->duration > 0)
lane.mediaEnd = lane.mediaStart + stream->duration * lane.tick;
if (slot == 0 && stream->codecpar->sample_rate > 0)
lane.preroll = std::max(stream->codecpar->seek_preroll, 0) /
static_cast<double>(stream->codecpar->sample_rate);
}
}
scratch = av_packet_alloc();
if (!scratch) throw std::bad_alloc();
reader = std::thread([this] { run(); });
}
void stop() {
cancelled.store(true, std::memory_order_relaxed);
budget.changed.notify_all();
if (reader.joinable()) reader.join();
}
void updatePosition(double seconds) {
if (!std::isfinite(seconds)) return;
std::lock_guard lock(budget.mutex);
position = std::max(0.0, seconds);
for (int slot = 0; slot < 2; ++slot) trim(slot);
budget.changed.notify_all();
}
// 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<double>(timestamp) / AV_TIME_BASE :
static_cast<double>(timestamp) * av_q2d(format->streams[streamIndex]->time_base)) - origin;
std::array<uint64_t, 2> cursors{};
bool cached = cacheSeeking && !seeking && !failure && std::isfinite(target);
for (int slot = 0; slot < 2 && cached; ++slot) {
const auto &lane = lanes[slot];
if (lane.index < 0 || (slot == 0 && skipAudio)) continue;
// A forced finish may have interrupted demux. Only a completely cached short
// 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, slot == 0 && std::isfinite(audioTarget) ? audioTarget : target, cursors[slot]);
}
position = std::max(0.0, target);
if (cached) {
for (int slot = 0; slot < 2; ++slot)
if (lanes[slot].index >= 0 && !(slot == 0 && skipAudio)) lanes[slot].cursor = cursors[slot];
// Cache hits never interrupt the outstanding read at the fetch frontier.
halted = false;
budget.changed.notify_all();
playerWake.notify_one();
return;
}
for (auto &lane : lanes) clear(lane);
audioDropped = false;
ended = halted = false;
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<int64_t>((earliest + origin) * AV_TIME_BASE);
}
seeking = true;
serial.fetch_add(1, std::memory_order_release);
budget.changed.notify_all();
}
void discardAudio(bool discard) {
std::lock_guard lock(budget.mutex);
skipAudio = discard;
// Stepping must not leave an unread audio lane blocking a combined demuxer.
if (discard) {
lanes[0].cursor = tail(lanes[0]);
trim(0);
}
budget.changed.notify_all();
}
void finish() {
std::lock_guard lock(budget.mutex);
for (auto &lane : lanes) lane.cursor = tail(lane);
halted = true;
seeking = false;
failure = 0;
serial.fetch_add(1, std::memory_order_release);
budget.changed.notify_all();
}
bool seekPending() const {
std::lock_guard lock(budget.mutex);
return seeking;
}
int error() const {
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;
const auto &lane = lanes[slot];
if (lane.points.empty() || (slot == 0 && audioDropped && !trackEnded(lane))) return NAN;
if (slot == 1 && (lane.packets.empty() || position > lane.packets.back().end + lane.tick)) return NAN;
const double start = lane.points.front().time + lane.preroll;
return std::max(0.0, lane.atBeginning ? std::min(start, lane.mediaStart) : start);
}
double bufferedEnd(int slot) const {
std::lock_guard lock(budget.mutex);
if (failure || seeking || (slot == 0 && skipAudio)) return 0;
const auto &lane = lanes[slot];
return lane.packets.empty() ? 0 : lane.packets.back().end;
}
// 1: packet, 0: not yet available, AVERROR_EOF: this lane is drained.
int take(int slot, AVPacket *destination, double &bufferedEnd) {
std::lock_guard lock(budget.mutex);
if (failure) return failure;
bufferedEnd = 0;
auto &lane = lanes[slot];
if (lane.cursor < tail(lane)) {
auto &packet = lane.packets[lane.cursor - lane.first];
const int result = av_packet_ref(destination, &packet.packet);
if (result < 0) return result;
bufferedEnd = packet.end;
++lane.cursor;
trim(slot);
budget.changed.notify_all();
return 1;
}
return ended || halted || trackEnded(lane) ? AVERROR_EOF : 0;
}
private:
struct Packet {
AVPacket packet{};
double end = 0;
std::size_t bytes = 0;
~Packet() { av_packet_unref(&packet); }
Packet() = default;
Packet(const Packet &) = delete;
Packet &operator=(const Packet &) = delete;
};
struct SeekPoint { double time; uint64_t sequence; };
struct Lane {
std::deque<Packet> packets;
std::deque<SeekPoint> points;
int index = -1;
uint64_t first = 0, cursor = 0;
std::size_t bytes = 0;
double tick = 0, frontierEnd = 0, mediaStart = 0, mediaEnd = NAN, preroll = 0;
bool atBeginning = false;
};
const std::atomic<bool> &stopping;
const std::atomic<uint64_t> &epoch;
const uint64_t activeEpoch;
std::condition_variable &playerWake;
StreamCacheBudget &budget;
std::array<Lane, 2> lanes;
AVPacket *scratch = nullptr;
std::atomic<bool> cancelled{false};
std::atomic<uint64_t> serial{0}, operationSerial{0};
std::atomic<int64_t> deadline{std::numeric_limits<int64_t>::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;
std::thread reader;
static int64_t now() {
return std::chrono::duration_cast<std::chrono::nanoseconds>(
std::chrono::steady_clock::now().time_since_epoch()).count();
}
static uint64_t tail(const Lane &lane) { return lane.first + lane.packets.size(); }
static bool trackEnded(const Lane &lane) {
return std::isfinite(lane.mediaEnd) && lane.frontierEnd + lane.tick * 0.5 >= lane.mediaEnd;
}
static double packetTime(const Lane &lane, const AVPacket *packet) {
const int64_t timestamp = packet->pts != AV_NOPTS_VALUE ? packet->pts : packet->dts;
return timestamp == AV_NOPTS_VALUE ? NAN : static_cast<double>(timestamp) * lane.tick;
}
std::size_t packetBytes(const AVPacket *packet) const {
if (packet->size < 0) return budget.limit + 1;
std::size_t bytes = sizeof(Packet) + sizeof(SeekPoint);
const auto payload = packet->buf ? packet->buf->size : static_cast<std::size_t>(packet->size);
if (payload > budget.limit || bytes > budget.limit - payload) return budget.limit + 1;
bytes += payload;
for (int i = 0; i < packet->side_data_elems; ++i) {
const auto size = packet->side_data[i].size;
if (size > budget.limit || sizeof(AVPacketSideData) > budget.limit - size ||
bytes > budget.limit - size - sizeof(AVPacketSideData)) return budget.limit + 1;
bytes += size + sizeof(AVPacketSideData);
}
return bytes;
}
double packetEnd(int slot, const AVPacket *packet) const {
if (packet->pts == AV_NOPTS_VALUE || (packet->flags & AV_PKT_FLAG_DISCARD)) return 0;
int64_t timestamp = packet->pts;
if (slot == 1) {
// PTS can jump to a future reference frame before its B-frames arrive.
if (packet->dts == AV_NOPTS_VALUE) return 0;
timestamp = std::min(timestamp, packet->dts);
}
const double tick = lanes[slot].tick;
const double end = static_cast<double>(timestamp) * tick - origin +
static_cast<double>(std::max<int64_t>(packet->duration, 0)) * tick;
return std::isfinite(end) && end > 0 ? end : 0;
}
void pop(Lane &lane) {
const auto bytes = lane.packets.front().bytes;
budget.used -= bytes;
lane.bytes -= bytes;
lane.packets.pop_front();
++lane.first;
lane.atBeginning = false;
while (!lane.points.empty() && lane.points.front().sequence < lane.first) lane.points.pop_front();
}
void clear(Lane &lane) {
budget.used -= lane.bytes;
lane.packets.clear();
lane.points.clear();
lane.first = lane.cursor = 0;
lane.bytes = 0;
lane.frontierEnd = 0;
lane.atBeginning = false;
}
static auto pointAt(const Lane &lane, double seconds) {
return std::upper_bound(lane.points.begin(), lane.points.end(), seconds,
[](double value, const SeekPoint &point) { return value < point.time; });
}
bool cachedCursor(int slot, double target, uint64_t &cursor) const {
const auto &lane = lanes[slot];
if (slot == 0 && std::isfinite(lane.mediaEnd) && target + lane.tick >= lane.mediaEnd) {
cursor = tail(lane);
return true;
}
if (slot == 0 && audioDropped && !trackEnded(lane)) return false;
if (lane.points.empty() || lane.packets.empty() ||
target > lane.packets.back().end + lane.tick) return false;
const auto after = pointAt(lane, target - lane.preroll);
if (after == lane.points.begin()) {
if (!lane.atBeginning) return false;
cursor = lane.points.front().sequence;
} else cursor = std::prev(after)->sequence;
return true;
}
void trim(int slot) {
auto &lane = lanes[slot];
uint64_t keep = lane.cursor;
if (cacheSeeking) {
const auto after = pointAt(lane, position - behindSeconds - lane.preroll);
if (after == lane.points.begin()) return;
keep = std::min(keep, std::prev(after)->sequence);
}
while (!lane.packets.empty() && lane.first < keep) pop(lane);
}
std::size_t laneLimit() const {
unsigned tracks = 0;
for (const auto *input : budget.inputs) {
if (!input) continue;
if (!input->started) { ++tracks; continue; }
for (int slot = 0; slot < 2; ++slot)
if (input->lanes[slot].index >= 0 && !(slot == 0 && input->skipAudio)) ++tracks;
}
// Each required track gets a reserved eighth; spare capacity remains borrowable.
return tracks > 1 ? budget.limit - budget.limit / 8 : budget.limit;
}
bool makeRoom(Lane &lane, std::size_t bytes) {
const auto quota = std::max(laneLimit(), bytes);
while (budget.used > budget.limit - bytes || lane.bytes > quota - bytes) {
Lane *oldest = nullptr;
double age = -std::numeric_limits<double>::infinity();
for (auto *input : budget.inputs) {
if (!input) continue;
for (auto &candidate : input->lanes) {
if (candidate.packets.empty() || candidate.first >= candidate.cursor) continue;
if (lane.bytes > quota - bytes && &candidate != &lane) continue;
const double distance = input->position - candidate.packets.front().end;
if (!oldest || distance > age) { oldest = &candidate; age = distance; }
}
}
if (!oldest) return false;
pop(*oldest);
}
return true;
}
bool needsRead() const {
for (int slot = 0; slot < 2; ++slot) {
const auto &lane = lanes[slot];
if (lane.index < 0 || (slot == 0 && skipAudio)) continue;
const double goal = std::isfinite(lane.mediaEnd) ?
std::min(position + aheadSeconds, lane.mediaEnd) : position + aheadSeconds;
if (lane.frontierEnd + lane.tick < goal) return true;
}
return false;
}
bool stoppingNow() const {
return cancelled.load(std::memory_order_relaxed) ||
stopping.load(std::memory_order_relaxed) || epoch.load(std::memory_order_relaxed) != activeEpoch;
}
void run() noexcept {
try {
while (!stoppingNow()) {
uint64_t current;
int index;
int64_t timestamp;
bool doSeek;
{
std::unique_lock lock(budget.mutex);
budget.changed.wait_for(lock, std::chrono::milliseconds(20), [this] {
return stoppingNow() || seeking || (!ended && !halted && needsRead());
});
if (stoppingNow()) return;
if (!seeking && (ended || halted || !needsRead())) continue;
current = serial.load(std::memory_order_acquire);
doSeek = seeking;
index = seekIndex;
timestamp = seekTimestamp;
}
operationSerial.store(current, std::memory_order_relaxed);
deadline.store(now() + 15'000'000'000LL, std::memory_order_relaxed);
if (doSeek) {
const int result = index >= 0 ?
av_seek_frame(format, index, timestamp, AVSEEK_FLAG_BACKWARD) :
avformat_seek_file(format, -1, std::numeric_limits<int64_t>::min(),
timestamp, timestamp, AVSEEK_FLAG_BACKWARD);
if (result >= 0 && format->pb) {
// Interrupted reads leave AVIO's error sticky after a successful seek.
format->pb->error = 0;
format->pb->eof_reached = 0;
}
std::lock_guard lock(budget.mutex);
if (serial.load(std::memory_order_acquire) != current) continue;
seeking = false;
if (result < 0) { failure = result; ended = true; }
playerWake.notify_one();
if (result < 0) continue;
}
const int result = av_read_frame(format, scratch);
std::unique_lock lock(budget.mutex);
if (stoppingNow()) return;
if (serial.load(std::memory_order_acquire) != current) {
av_packet_unref(scratch);
continue;
}
if (result == AVERROR(EAGAIN)) {
budget.changed.wait_for(lock, std::chrono::milliseconds(5));
continue;
}
if (result < 0) {
av_packet_unref(scratch);
ended = true;
if (result != AVERROR_EOF) failure = result;
playerWake.notify_one();
continue;
}
int slot = -1;
for (int i = 0; i < 2; ++i)
if (lanes[i].index >= 0 && scratch->stream_index == lanes[i].index) slot = i;
if (slot < 0 || (slot == 0 && skipAudio)) {
if (slot == 0) audioDropped = true;
av_packet_unref(scratch);
continue;
}
auto &lane = lanes[slot];
const auto bytes = packetBytes(scratch);
if (bytes > budget.limit) {
failure = scratch->size < 0 ? AVERROR_INVALIDDATA : AVERROR(ENOBUFS);
ended = true;
av_packet_unref(scratch);
playerWake.notify_one();
continue;
}
// The demuxer's single in-flight packet is not retained until it fits.
// Byte pressure shortens consumed history, never drops unread packets.
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;
av_packet_unref(scratch);
continue;
}
const double time = packetTime(lane, scratch) - origin;
const double end = std::max(lane.frontierEnd, packetEnd(slot, scratch));
const uint64_t sequence = tail(lane);
lane.packets.emplace_back();
auto &packet = lane.packets.back();
packet.bytes = bytes;
packet.end = end;
av_packet_move_ref(&packet.packet, scratch);
lane.bytes += bytes;
budget.used += bytes;
lane.frontierEnd = end;
const bool seekPoint = !(packet.packet.flags & AV_PKT_FLAG_DISCARD) &&
(slot == 0 || (packet.packet.flags & AV_PKT_FLAG_KEY));
if (seekPoint && std::isfinite(time) &&
(lane.points.empty() || time >= lane.points.back().time)) {
lane.points.push_back({time, sequence});
if (lane.points.size() == 1 && sequence == lane.first &&
time <= lane.mediaStart + lane.tick) lane.atBeginning = true;
}
trim(slot);
playerWake.notify_one();
}
} catch (...) {
std::lock_guard lock(budget.mutex);
failure = AVERROR(ENOMEM);
ended = true;
playerWake.notify_one();
}
}
};
}
Reference in New Issue View Git Blame Copy Permalink