508 lines
22 KiB
C++
508 lines
22 KiB
C++
#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();
|
|
}
|
|
void seek(int streamIndex, int64_t timestamp) {
|
|
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, 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;
|
|
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;
|
|
}
|
|
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;
|
|
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.
|
|
budget.changed.wait(lock, [&] {
|
|
return stoppingNow() || serial.load(std::memory_order_acquire) != current ||
|
|
(slot == 0 && skipAudio) || makeRoom(lane, bytes);
|
|
});
|
|
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();
|
|
}
|
|
}
|
|
};
|
|
}
|