server/src/pyrowave_policy.cpp
Rama main · réplica de lectura del repositorio original.
/**
* @file src/pyrowave_policy.cpp
* @brief Pure PyroWave host policy: frame framing and the per-frame byte budget.
*/
#include "pyrowave_policy.h"
#include "pyrowave_protocol.h"
#include <algorithm>
#include <cstring>
#include <limits>
#include <stdexcept>
namespace pyrowave::policy {
namespace {
using namespace pyrowave::protocol;
std::uint32_t read_u32(const std::uint8_t *p) {
return std::uint32_t(p[0]) | (std::uint32_t(p[1]) << 8) | (std::uint32_t(p[2]) << 16) | (std::uint32_t(p[3]) << 24);
}
void append_u32(std::vector<std::uint8_t> &out, std::uint32_t value) {
out.push_back(std::uint8_t(value & 0xff));
out.push_back(std::uint8_t((value >> 8) & 0xff));
out.push_back(std::uint8_t((value >> 16) & 0xff));
out.push_back(std::uint8_t((value >> 24) & 0xff));
}
void append_padding(std::vector<std::uint8_t> &out, std::size_t bytes) {
// bytes >= PADDING_RECORD_MIN_BYTES and a multiple of four.
append_u32(out, PADDING_MAGIC);
append_u32(out, std::uint32_t(bytes / 4 - 2));
out.resize(out.size() + (bytes - PADDING_RECORD_MIN_BYTES), 0);
}
struct record_t {
std::size_t offset;
std::size_t size;
std::uint32_t block_index;
};
/// Records too large to share a shard with a padding record. A record of a shard
/// less 4 bytes would always strand a 4-byte remainder, so it counts too.
bool oversized(std::size_t record_bytes, std::size_t shard_payload) {
return record_bytes + PADDING_RECORD_MIN_BYTES > shard_payload;
}
/// Bytes left in the current shard at frame byte `position` (never 0).
std::size_t shard_remaining(std::size_t position, std::size_t shard_payload) {
return shard_payload - (position + FRAME_HEADER_BYTES) % shard_payload;
}
/**
* Finds the earliest remaining record no larger than a limit, in O(log) time.
*
* Records are bucketed by size in words (payload_words is 12 bits), each bucket
* holding its record indices in frame order. A min segment tree over the buckets
* keeps each bucket's earliest remaining index, so "earliest record of at most N
* words" is a prefix-minimum query. Records are only ever removed from the front of
* their bucket: both the query and the frame-order cursor pick a bucket's earliest.
*/
class fit_index_t {
public:
static constexpr std::size_t NONE = std::numeric_limits<std::size_t>::max();
explicit fit_index_t(const std::vector<std::uint32_t> &words):
heads(LEAVES + 1, 0),
ends(LEAVES, 0),
order(words.size()),
tree(2 * LEAVES, NONE) {
// Counting sort of the record indices by size, stable in frame order.
for (const auto w : words) {
++heads[w + 1];
}
for (std::size_t b = 0; b < LEAVES; b++) {
heads[b + 1] += heads[b];
ends[b] = heads[b + 1];
}
std::vector<std::size_t> fill(heads.begin(), heads.end() - 1);
for (std::size_t i = 0; i < words.size(); i++) {
order[fill[words[i]]++] = i;
}
for (std::size_t b = 0; b < LEAVES; b++) {
tree[LEAVES + b] = heads[b] < ends[b] ? order[heads[b]] : NONE;
}
for (std::size_t n = LEAVES - 1; n > 0; n--) {
tree[n] = std::min(tree[2 * n], tree[2 * n + 1]);
}
}
/// Earliest remaining record of at most `max_words` words, or NONE.
[[nodiscard]] std::size_t earliest_fitting(std::size_t max_words) const {
if (max_words == 0) {
return NONE;
}
std::size_t best = NONE;
std::size_t lo = LEAVES;
std::size_t hi = LEAVES + std::min(max_words, LEAVES - 1) + 1;
while (lo < hi) {
if (lo & 1) {
best = std::min(best, tree[lo++]);
}
if (hi & 1) {
best = std::min(best, tree[--hi]);
}
lo >>= 1;
hi >>= 1;
}
return best;
}
/// Earliest remaining record of exactly `words` words, or NONE.
[[nodiscard]] std::size_t earliest_exact(std::uint32_t words) const {
return words < LEAVES ? tree[LEAVES + words] : NONE;
}
/// Remove the earliest remaining record of `words` words.
void remove(std::uint32_t words) {
const auto head = ++heads[words];
std::size_t n = LEAVES + words;
tree[n] = head < ends[words] ? order[head] : NONE;
for (n >>= 1; n > 0; n >>= 1) {
tree[n] = std::min(tree[2 * n], tree[2 * n + 1]);
}
}
private:
static constexpr std::size_t LEAVES = 4096;
std::vector<std::size_t> heads; ///< Per size: position of the earliest remaining record in `order`.
std::vector<std::size_t> ends; ///< Per size: end of its run in `order`.
std::vector<std::size_t> order; ///< Record indices grouped by size, in frame order.
std::vector<std::size_t> tree;
};
} // namespace
std::uint32_t coarse_block_count(std::uint32_t width, std::uint32_t height) {
// PyroWave pads each dimension to 32 pixels (at least 128), then five wavelet
// levels halve it; the coarsest level holds four bands of three components.
const auto blocks = [](std::uint32_t extent) {
const std::uint32_t aligned = std::max<std::uint32_t>((extent + 31) / 32 * 32, 128);
return ((aligned >> 5) + 31) / 32;
};
return blocks(width) * blocks(height) * 4 * 3;
}
framing_e select_framing(bool client_sent_adaptive_fec, std::optional<std::uint32_t> client_features) {
if (client_sent_adaptive_fec) {
return framing_e::records;
}
if (client_features && (*client_features & FEATURE_RECORD_FRAMING)) {
return framing_e::records;
}
return framing_e::length_prefixed;
}
std::size_t shard_payload_bytes(int packetsize) {
if (packetsize <= SHARD_OVERHEAD_BYTES) {
return 0;
}
const auto shard = std::size_t(packetsize - SHARD_OVERHEAD_BYTES);
if (shard < FRAME_HEADER_BYTES + SEQUENCE_HEADER_BYTES + PADDING_RECORD_MIN_BYTES || shard % 4 != 0) {
return 0;
}
return shard;
}
std::optional<record_frame_stats_t> write_record_frame(
std::span<const std::uint8_t> bitstream,
std::size_t shard_payload,
std::vector<std::uint8_t> &out,
std::size_t frame_limit
) {
// Split the frame into its sequence header and block records.
if (bitstream.size() < 8 || (read_u32(bitstream.data()) >> 31) == 0 ||
(frame_limit && bitstream.size() > frame_limit)) {
return std::nullopt;
}
std::vector<record_t> records;
for (std::size_t position = 8; position < bitstream.size();) {
if (bitstream.size() - position < 8) {
return std::nullopt;
}
const std::uint32_t word0 = read_u32(bitstream.data() + position);
const std::uint32_t payload_words = (word0 >> 16) & 0xfff;
if ((word0 >> 31) != 0 || payload_words < 2 || std::size_t(payload_words) * 4 > bitstream.size() - position) {
return std::nullopt;
}
records.push_back({position, std::size_t(payload_words) * 4, read_u32(bitstream.data() + position + 4) >> 8});
position += records.back().size;
}
record_frame_stats_t stats;
stats.block_records = records.size();
const std::size_t frame_start = out.size();
out.reserve(frame_start + bitstream.size());
out.insert(out.end(), bitstream.begin(), bitstream.begin() + 8);
if (!shard_payload) {
out.insert(out.end(), bitstream.begin() + 8, bitstream.end());
return stats;
}
const std::uint32_t sequence_word = read_u32(bitstream.data());
const std::uint32_t coarse_blocks = coarse_block_count((sequence_word & 0x3fff) + 1, ((sequence_word >> 14) & 0x3fff) + 1);
// Frame byte p sits at offset p + FRAME_HEADER_BYTES of the shard stream,
// because the short frame header precedes the frame in the first shard.
const auto remaining = [&]() {
return shard_remaining(out.size() - frame_start, shard_payload);
};
const auto place = [&](const record_t &record) {
out.insert(out.end(), bitstream.begin() + record.offset, bitstream.begin() + record.offset + record.size);
};
const auto pad = [&](std::size_t bytes) {
append_padding(out, bytes);
++stats.padding_records;
stats.padding_bytes += bytes;
};
// The coarsest wavelet level goes first: a receiver cannot decode a frame that
// lost any of it, so those shards (critical_bytes) are the ones worth parity.
// Then everything else. Within each group, oversized records come first and span
// shards wherever they start; each is shifted by a minimal padding record when it
// would end 4 bytes before a boundary, since nothing could fill that remainder.
// The group's other records follow, packed first-fit without crossing a shard
// boundary, so every shard after the finer level's oversized records starts with
// a record.
std::vector<std::size_t> coarse;
std::vector<std::size_t> fine;
for (std::size_t i = 0; i < records.size(); i++) {
(records[i].block_index < coarse_blocks ? coarse : fine).push_back(i);
}
for (const auto *group : {&coarse, &fine}) {
std::vector<std::size_t> ordinary;
for (const auto i : *group) {
if (oversized(records[i].size, shard_payload)) {
if (records[i].size % shard_payload == remaining() - 4) {
pad(PADDING_RECORD_MIN_BYTES);
}
place(records[i]);
++stats.oversized_records;
} else {
ordinary.push_back(i);
}
}
std::vector<std::uint32_t> words;
words.reserve(ordinary.size());
for (const auto i : ordinary) {
words.push_back(std::uint32_t(records[i].size / 4));
}
fit_index_t fit(words);
std::vector<bool> placed(ordinary.size(), false);
std::size_t earliest = 0;
for (std::size_t left = ordinary.size(); left != 0;) {
while (placed[earliest]) {
++earliest;
}
const std::size_t room = remaining();
std::size_t chosen = fit.earliest_fitting(room / 4);
if (chosen != fit_index_t::NONE && room - std::size_t(words[chosen]) * 4 == 4) {
// That would strand 4 bytes, too few for a padding record. Take a record
// that fills the shard exactly or leaves room for a padding record.
const std::size_t exact = fit.earliest_exact(std::uint32_t(room / 4));
const std::size_t roomy = room >= 16 ? fit.earliest_fitting((room - 8) / 4) : fit_index_t::NONE;
chosen = std::min(exact, roomy);
}
if (chosen == fit_index_t::NONE) {
// Never 4: shards start with at least 64 bytes and no placement leaves 4.
pad(room);
continue;
}
if (chosen != earliest) {
++stats.reordered_records;
}
placed[chosen] = true;
fit.remove(words[chosen]);
--left;
place(records[ordinary[chosen]]);
}
if (group == &coarse) {
stats.critical_bytes = out.size() - frame_start;
}
}
if (frame_limit && out.size() - frame_start > frame_limit) {
out.resize(frame_start);
out.insert(out.end(), bitstream.begin(), bitstream.end());
stats = {};
stats.block_records = records.size();
}
return stats;
}
void write_length_prefixed_frame(
std::span<const packet_t> packets,
const std::uint8_t *bitstream,
std::vector<std::uint8_t> &out
) {
std::size_t total = 4;
for (const auto &packet : packets) {
total += 4 + packet.size;
}
out.reserve(out.size() + total);
append_u32(out, std::uint32_t(packets.size()));
for (const auto &packet : packets) {
append_u32(out, std::uint32_t(packet.size));
out.insert(out.end(), bitstream + packet.offset, bitstream + packet.offset + packet.size);
}
}
record_frame_info_t inspect_record_frame(std::span<const std::uint8_t> frame, std::size_t shard_payload) {
record_frame_info_t info;
std::size_t position = 0;
std::uint32_t last_block_index = 0;
auto fail = [&](std::string error) {
info.valid = false;
info.error = std::move(error);
return info;
};
std::uint32_t coarse_blocks = 0;
bool seen_ordinary_coarse = false;
bool seen_ordinary_fine = false;
bool seen_fine = false;
// Returns whether the record is oversized; ordinary records and padding must
// stay inside one shard so that every shard after the oversized records
// starts with a record.
auto note_alignment = [&](std::size_t start, std::size_t size) {
if (!shard_payload) {
return false;
}
const std::size_t wire_start = start + FRAME_HEADER_BYTES;
const std::size_t first_shard = wire_start / shard_payload;
const std::size_t last_shard = (wire_start + size - 1) / shard_payload;
if (oversized(size, shard_payload)) {
++info.oversized_records;
return true;
}
if (first_shard != last_shard) {
++info.misaligned_records;
}
return false;
};
info.stripped.reserve(frame.size());
while (position < frame.size()) {
if (frame.size() - position < 8) {
return fail("truncated record header at byte " + std::to_string(position));
}
const auto *record = frame.data() + position;
const std::uint32_t word0 = read_u32(record);
const std::uint32_t word1 = read_u32(record + 4);
if (word0 == PADDING_MAGIC) {
const std::size_t bytes = PADDING_RECORD_MIN_BYTES + std::size_t(word1) * 4;
if (bytes > frame.size() - position) {
return fail("padding record runs past the end of the frame at byte " + std::to_string(position));
}
++info.padding_records;
info.padding_bytes += bytes;
note_alignment(position, bytes);
position += bytes;
continue;
}
const bool extended = (word0 >> 31) != 0;
if (extended) {
if (info.has_sequence_header) {
return fail("second sequence header at byte " + std::to_string(position));
}
if (info.block_records != 0) {
return fail("sequence header after block records");
}
info.has_sequence_header = true;
info.width = (word0 & 0x3fff) + 1;
info.height = ((word0 >> 14) & 0x3fff) + 1;
info.total_blocks = word1 & 0xffffff;
const std::uint32_t code = (word1 >> 24) & 0x3;
if (code != 0) {
return fail("sequence header code " + std::to_string(code) + " is not START_OF_FRAME");
}
info.chroma444 = ((word1 >> 26) & 0x1) != 0;
coarse_blocks = coarse_block_count(info.width, info.height);
info.critical_bytes = position + 8;
note_alignment(position, 8);
info.stripped.insert(info.stripped.end(), record, record + 8);
position += 8;
continue;
}
if (!info.has_sequence_header) {
return fail("block record before the sequence header");
}
const std::uint32_t payload_words = (word0 >> 16) & 0xfff;
if (payload_words < 2) {
return fail("block record with payload_words " + std::to_string(payload_words) + " at byte " + std::to_string(position));
}
const std::size_t bytes = std::size_t(payload_words) * 4;
if (bytes > frame.size() - position) {
return fail("block record runs past the end of the frame at byte " + std::to_string(position));
}
const std::uint32_t block_index = word1 >> 8;
if (info.block_records != 0 && block_index < last_block_index) {
++info.out_of_order_records;
}
last_block_index = block_index;
++info.block_records;
if (shard_payload) {
const bool coarse = block_index < coarse_blocks;
if (coarse && seen_fine) {
++info.late_coarse_records;
}
if (note_alignment(position, bytes)) {
// Each level's oversized records precede its ordinary ones
if (coarse ? seen_ordinary_coarse : seen_ordinary_fine) {
++info.late_oversized_records;
}
} else if (coarse) {
seen_ordinary_coarse = true;
} else {
seen_ordinary_fine = true;
}
if (!coarse) {
seen_fine = true;
} else {
info.critical_bytes = position + bytes;
}
}
info.stripped.insert(info.stripped.end(), record, record + bytes);
position += bytes;
}
if (!info.has_sequence_header) {
return fail("frame has no sequence header");
}
info.valid = true;
return info;
}
std::vector<bool> record_start_shards(std::span<const std::uint8_t> frame, std::size_t shard_payload) {
if (!shard_payload || frame.empty()) {
return {};
}
std::vector<bool> starts((frame.size() + FRAME_HEADER_BYTES + shard_payload - 1) / shard_payload, false);
for (std::size_t position = 0; position < frame.size();) {
if (frame.size() - position < 8) {
return {};
}
if ((position + FRAME_HEADER_BYTES) % shard_payload == 0 || position == 0) {
starts[(position + FRAME_HEADER_BYTES) / shard_payload] = true;
}
const std::uint32_t word0 = read_u32(frame.data() + position);
std::size_t bytes;
if (word0 == PADDING_MAGIC) {
bytes = PADDING_RECORD_MIN_BYTES + std::size_t(read_u32(frame.data() + position + 4)) * 4;
} else if ((word0 >> 31) != 0) {
bytes = 8;
} else {
bytes = std::size_t((word0 >> 16) & 0xfff) * 4;
if (bytes < 8) {
return {};
}
}
if (bytes > frame.size() - position) {
return {};
}
position += bytes;
}
return starts;
}
std::size_t parity_shards(std::size_t data_shards, int fec_percentage, std::size_t min_parity_shards) {
if (fec_percentage <= 0) {
return 0;
}
return std::max((data_shards * std::size_t(fec_percentage) + 99) / 100, min_parity_shards);
}
std::vector<fec_block_t> plan_fec_blocks(std::size_t total_shards, std::size_t critical_shards, int critical_fec_percentage, std::size_t min_parity_shards, int detail_fec_percentage, std::size_t extra_parity_budget) {
std::vector<fec_block_t> blocks;
if (total_shards == 0) {
return blocks;
}
// Blocks without FEC are kept to Reed-Solomon size too, as for the other codecs,
// so a block that stops arriving is delivered without waiting for a huge one.
const auto split = [&](std::size_t shards, std::size_t max_blocks) {
const std::size_t count = std::clamp<std::size_t>((shards + MAX_REED_SOLOMON_SHARDS - 1) / MAX_REED_SOLOMON_SHARDS, 1, max_blocks);
const std::size_t per_block = (shards + count - 1) / count;
for (std::size_t i = 0; i < count; i++) {
blocks.push_back({i + 1 < count ? per_block : shards - per_block * (count - 1), 0});
}
};
const std::size_t critical = std::min(critical_shards, total_shards);
const bool protect = critical != 0 && critical_fec_percentage > 0 &&
critical + parity_shards(critical, critical_fec_percentage, min_parity_shards) <= MAX_REED_SOLOMON_SHARDS;
if (!protect) {
split(total_shards, MAX_FEC_BLOCKS);
return blocks;
}
blocks.push_back({critical, critical_fec_percentage});
detail_fec_percentage = std::clamp(detail_fec_percentage, 0, MAX_DETAIL_FEC_PERCENTAGE);
if (detail_fec_percentage > 0 && extra_parity_budget > 0 && total_shards > critical) {
const auto fine = total_shards - critical;
// Bound the total extra overhead even when only part of a large frame can
// be protected. Never spend credit accumulated across an idle period.
auto budget = std::min(extra_parity_budget, parity_shards(fine, detail_fec_percentage, 0));
const auto capacity = [&](int percentage) {
std::size_t data = MAX_REED_SOLOMON_SHARDS;
while (data && data + parity_shards(data, percentage, min_parity_shards) > MAX_REED_SOLOMON_SHARDS) {
--data;
}
return data;
};
// Prefer covering all detail, reducing the rate if needed. Even splitting
// avoids leaving the last block with a disproportionately high minimum FEC.
for (int percentage = detail_fec_percentage; percentage > 0; --percentage) {
const auto cap = capacity(percentage);
if (!cap) {
continue;
}
const auto count = (fine + cap - 1) / cap;
if (count > MAX_FEC_BLOCKS - 1) {
continue;
}
std::vector<fec_block_t> candidate = blocks;
std::size_t parity = 0;
for (std::size_t i = 0; i < count; ++i) {
const auto data = fine / count + (i < fine % count ? 1 : 0);
candidate.push_back({data, percentage});
parity += parity_shards(data, percentage, min_parity_shards);
}
if (parity <= budget) {
return candidate;
}
}
// Large frames cannot fit entirely in Reed-Solomon blocks. Extend the
// critical block first, then protect additional prefixes only when enough
// block slots remain to carry every unprotected shard.
const auto base_parity = parity_shards(critical, critical_fec_percentage, min_parity_shards);
auto leading = critical;
const auto leading_cap = std::min(total_shards, capacity(critical_fec_percentage));
while (leading < leading_cap && parity_shards(leading + 1, critical_fec_percentage, min_parity_shards) - base_parity <= budget) {
++leading;
}
blocks.front().data_shards = leading;
budget -= parity_shards(leading, critical_fec_percentage, min_parity_shards) - base_parity;
auto remaining = total_shards - leading;
const auto cap = capacity(detail_fec_percentage);
while (remaining && blocks.size() < MAX_FEC_BLOCKS && cap) {
auto data = std::min(remaining, cap);
while (data && parity_shards(data, detail_fec_percentage, min_parity_shards) > budget) {
--data;
}
const auto slots_after = MAX_FEC_BLOCKS - blocks.size() - 1;
if (!data || remaining - data > slots_after * MAX_FEC_BLOCK_SHARDS) {
break;
}
blocks.push_back({data, detail_fec_percentage});
budget -= parity_shards(data, detail_fec_percentage, min_parity_shards);
remaining -= data;
}
if (remaining) {
split(remaining, MAX_FEC_BLOCKS - blocks.size());
}
return blocks;
}
if (total_shards > critical) {
split(total_shards - critical, MAX_FEC_BLOCKS - 1);
}
return blocks;
}
detail_fec_controller_t::detail_fec_controller_t(int framerate):
nominal_interval {framerate > 0 ? 1.0 / framerate : 0.0} {
if (framerate <= 0) {
throw std::invalid_argument("PyroWave requires a negotiated frame rate");
}
}
detail_fec_t detail_fec_controller_t::observe(std::span<const std::uint8_t> frame, std::chrono::steady_clock::time_point when, int bitrate_kbps) {
const auto reset = [&]() -> detail_fec_t {
previous_blocks.clear();
previous_record_bytes = 0;
previous_time.reset();
interval = stable_seconds = 0.0;
return {};
};
const auto elapsed = previous_time ? std::chrono::duration<double>(when - *previous_time).count() : 0.0;
if (elapsed > 0.0 && elapsed <= nominal_interval * 1.01) {
// At negotiated cadence, allow 1% timing noise without an activity scan.
// Start a fresh similarity history if cadence subsequently falls.
previous_blocks.clear();
previous_record_bytes = 0;
previous_time = when;
interval = elapsed;
stable_seconds = 0.0;
return {};
}
if (frame.size() < SEQUENCE_HEADER_BYTES) {
return reset();
}
const auto word0 = read_u32(frame.data());
const auto word1 = read_u32(frame.data() + 4);
const auto count = word1 & 0xffffff;
if (!(word0 & 0x80000000u) || word0 == PADDING_MAGIC || ((word1 >> 24) & 3) != 0 || !count || count > (frame.size() - 8) / 8) {
return reset();
}
// Sequence and nonzero block count can change without changing the format.
// Zero coefficient blocks are omitted, so block IDs need not be contiguous.
const auto format = (std::uint64_t(word0 & 0x8fffffffu) << 32) | (word1 & 0xff000000u);
std::unordered_map<std::uint32_t, std::uint64_t> hashes;
hashes.reserve(count);
std::size_t unchanged_bytes = 0;
std::size_t record_bytes = 0;
std::size_t records = 0;
for (std::size_t position = 8; position < frame.size();) {
if (frame.size() - position < 8) {
return reset();
}
const auto *record = frame.data() + position;
const auto header = read_u32(record);
const auto index_word = read_u32(record + 4);
const bool padding = header == PADDING_MAGIC;
const auto bytes = padding ? 8 + std::size_t(index_word) * 4 : std::size_t((header >> 16) & 0xfff) * 4;
if (bytes < 8 || bytes > frame.size() - position || (!padding && (header >> 31))) {
return reset();
}
if (!padding) {
const auto index = index_word >> 8;
// Hash complete records, masking only the three sequence bits. Indexing
// by block ID makes padding and the record packer's reordering irrelevant.
std::uint64_t hash = 14695981039346656037ull;
for (std::size_t i = 0; i < bytes; ++i) {
hash ^= i == 3 ? record[i] & 0x8f : record[i];
hash *= 1099511628211ull;
}
if (!hashes.emplace(index, hash).second) {
return reset();
}
const auto previous = previous_blocks.find(index);
if (format == previous_format && previous != previous_blocks.end() && hash == previous->second) {
unchanged_bytes += bytes;
}
record_bytes += bytes;
++records;
}
position += bytes;
}
if (records != count) {
return reset();
}
const bool similar = double(unchanged_bytes) >= 0.75 * double(std::max(record_bytes, previous_record_bytes));
previous_blocks = std::move(hashes);
previous_record_bytes = record_bytes;
previous_format = format;
previous_time = when;
if (elapsed <= 0.0 || elapsed > 0.25) {
interval = stable_seconds = 0.0;
return {};
}
interval = interval > 0.0 ? interval + (elapsed / (0.25 + elapsed)) * (elapsed - interval) : elapsed;
// Require both current and smoothed cadence to be slow. A single hitch must
// not enable parity; a return to negotiated FPS turns it off immediately.
const auto cadence = std::min(interval, elapsed);
stable_seconds = similar && cadence > nominal_interval * 1.01 ? stable_seconds + std::min(elapsed, 0.05) : 0.0;
if (stable_seconds < 0.25) {
return {};
}
detail_fec_t result;
// Stable encoded frame size frees this fraction of a frame's payload budget
// as cadence falls. The sender deducts wire overhead before spending it.
result.percentage = int(std::clamp(100.0 * (cadence / nominal_interval - 1.0), 0.0, double(MAX_DETAIL_FEC_PERCENTAGE)));
// Full wire cost (including base FEC) is deducted by the sender. This is a
// per-frame allowance, not a token bucket; stalls cannot fund a large burst.
result.frame_wire_budget = std::size_t(double(std::max(bitrate_kbps, 0)) * 125.0 * std::min(cadence, 0.05));
return result;
}
std::size_t max_frame_bytes(int packetsize, bool critical_fec) {
if (packetsize <= SHARD_OVERHEAD_BYTES) {
return 0;
}
const auto max_shards = critical_fec ? 3000u : 4000u;
return std::min<std::size_t>(max_shards, protocol::MAX_FEC_BLOCKS * protocol::MAX_DATA_SHARDS_PER_BLOCK) *
std::size_t(packetsize - SHARD_OVERHEAD_BYTES) - FRAME_HEADER_BYTES;
}
std::size_t max_bitstream_bytes(int packetsize, bool length_prefixed, bool critical_fec) {
auto capacity = max_frame_bytes(packetsize, critical_fec);
if (length_prefixed) {
// At worst each block and the sequence header occupy their own packet.
// Each packet needs a length word, plus the frame's packet-count word.
if (capacity < sizeof(std::uint32_t)) {
return 0;
}
capacity = (capacity - sizeof(std::uint32_t)) /
(BLOCK_HEADER_BYTES + sizeof(std::uint32_t)) * BLOCK_HEADER_BYTES;
}
return capacity & ~std::size_t(3);
}
std::size_t pacing_packets_per_ms(
std::uint64_t link_bps, int bitrate_kbps,
std::size_t payload_bytes, std::size_t wire_bytes,
std::size_t frame_bytes, int framerate
) {
if (payload_bytes == 0 || wire_bytes == 0) {
return 1;
}
// Link capacity includes network headers; the negotiated bitrate is payload.
const auto frame_packets = (frame_bytes + payload_bytes - 1) / payload_bytes;
const auto frame_allowance = (frame_packets * std::uint64_t(std::max(framerate, 0)) + 999) / 1000;
const auto bitrate_allowance = (std::uint64_t(std::max(bitrate_kbps, 0)) + 8 * payload_bytes - 1) / (8 * payload_bytes);
// Round demand up so quantizing to a pacing quantum cannot create a backlog.
// Round link capacity down so pacing never exceeds the reported link speed.
const auto packets = link_bps != 0 ? link_bps / 8 / 1000 / wire_bytes :
std::max(frame_allowance, bitrate_allowance);
return std::max<std::size_t>(1, packets);
}
budget_t::budget_t(int framerate, int bitrate_kbps, std::size_t max_frame_bytes, bool stable_frame_size):
bitrate_kbps {bitrate_kbps},
max_frame_bytes {max_frame_bytes},
nominal_interval {framerate > 0 ? 1.0 / framerate : 0.0},
stable_frame_size {stable_frame_size},
frame_interval {nominal_interval} {
if (framerate <= 0) {
throw std::invalid_argument("PyroWave requires a negotiated frame rate");
}
update();
}
void budget_t::set_bitrate(int kbps) {
bitrate_kbps = kbps;
update();
}
void budget_t::on_frame(std::chrono::steady_clock::time_point when) {
if (last_frame) {
if (when <= *last_frame) {
budget = 0;
return;
}
frame_interval = std::chrono::duration<double>(when - *last_frame).count();
}
last_frame = when;
update();
}
void budget_t::update() {
const auto budget_interval = stable_frame_size ? std::min(frame_interval, nominal_interval) : frame_interval;
double bytes = std::max(0.0, double(bitrate_kbps) * 1000.0 * budget_interval / 8.0);
if (max_frame_bytes) {
bytes = std::min(bytes, double(max_frame_bytes));
}
bytes = std::min(bytes, double(std::numeric_limits<std::uint32_t>::max() & ~3u));
budget = std::size_t(bytes) & ~std::size_t(3);
if (budget < SEQUENCE_HEADER_BYTES + BLOCK_HEADER_BYTES) {
budget = 0;
}
}
} // namespace pyrowave::policy