diff --git a/cpp/obs/src/moq-output.cpp b/cpp/obs/src/moq-output.cpp index a10a03ca551..29ad0eaab46 100644 --- a/cpp/obs/src/moq-output.cpp +++ b/cpp/obs/src/moq-output.cpp @@ -384,9 +384,9 @@ void MoQOutput::Reset() for (auto &[encoder, track] : video_tracks) { if (track.producer) track.producer->finish(); - if (track.rate && track.rate->Target() != track.rate->Max()) { + if (track.rate && track.applied_bps != track.bitrate_bps) { if (OBSEncoder strong = OBSGetStrongRef(track.encoder)) - restore.emplace_back(std::move(strong), track.rate->Max()); + restore.emplace_back(std::move(strong), track.bitrate_bps); } } video_tracks.clear(); @@ -567,8 +567,8 @@ bool MoQOutput::TryGetVideoRate(VideoRate *out) if (!track.rate || !track.reservation) continue; following = true; - rate.target_bps += track.rate->Target(); - rate.configured_bps += track.rate->Max(); + rate.target_bps += track.enabled ? track.applied_bps : 0; + rate.configured_bps += track.bitrate_bps; } } if (following && out) @@ -657,6 +657,9 @@ void MoQOutput::VideoData(struct encoder_packet *packet) auto it = video_tracks.find(encoder); if (it == video_tracks.end()) { + // A track opens on a keyframe; deltas before it have no group to join. + if (!packet->keyframe) + return; VideoInit(encoder); it = video_tracks.find(encoder); } @@ -668,24 +671,74 @@ void MoQOutput::VideoData(struct encoder_packet *packet) // Read before the frame is written, so a retune lands on the encoder's next // frame. obs_encoder_update only flags an active encoder for its own thread. if (track.rate && track.reservation) { - if (auto next = track.rate->Update(track.reservation->rate(), MoQRate::Clock::now())) - SetBitrate(encoder, *next); + if (auto next = track.rate->update(track.reservation)) + Retarget(encoder, track, *next); } + // OBS can't stop one encoder of a running output, or force it to keyframe, so a + // disabled rung keeps encoding and its packets are dropped here. It resumes on + // the encoder's next natural keyframe once the grant funds it again. + if (!track.sending && (!track.enabled || !packet->keyframe)) + return; + auto timestamp = PacketTimestamp(packet); if (!timestamp) { LOG_WARNING("Dropping video frame with negative PTS: %lld", (long long)packet->pts); return; } + // A resume that fails to write its keyframe stays disabled, so it retries on the + // next keyframe instead of advertising a rung whose deltas have no group to join. + bool resumed = false; + if (!track.sending) { + if (auto result = track.producer->set_enabled(true); !result) { + LOG_ERROR("Failed to enable video track: %s", result.error().to_string().c_str()); + return; + } + track.sending = true; + resumed = true; + } + if (auto result = track.producer->write_frame(PacketFrame(packet, *timestamp)); !result) { LOG_ERROR("Failed to write video frame: %s", result.error().to_string().c_str()); + if (resumed) { + track.sending = false; + if (auto disable = track.producer->set_enabled(false); !disable) + LOG_ERROR("Failed to disable video track: %s", disable.error().to_string().c_str()); + } return; } + track.started = true; total_bytes_sent += packet->size; } +// NOTE: Caller must hold media_mutex. +void MoQOutput::Retarget(obs_encoder_t *encoder, Track &track, const moq::RateTarget &target) +{ + if (const auto *enabled = std::get_if(&target.get_variant())) { + SetBitrate(encoder, enabled->bitrate_bps); + track.applied_bps = enabled->bitrate_bps; + track.enabled = true; + return; + } + + track.enabled = false; + if (!track.sending) + return; + track.sending = false; + // The lower renditions keep their share; viewers switch to one instead of + // waiting on frames that stopped coming. Before the first frame there is no + // timeline to break, and the marker would drop the init's parameter sets, which + // an encoder that sends them only as extra data never repeats. + if (track.started) { + if (auto result = track.producer->discontinuity(); !result) + LOG_ERROR("Failed to mark a disabled video track: %s", result.error().to_string().c_str()); + } + if (auto result = track.producer->set_enabled(false); !result) + LOG_ERROR("Failed to disable video track: %s", result.error().to_string().c_str()); +} + // NOTE: Caller must hold media_mutex. void MoQOutput::Reserve(Track &track) { @@ -757,6 +810,51 @@ void MoQOutput::VideoInit(obs_encoder_t *encoder) hint.optimize_for_latency = true; config.hint = hint; + // Every video encoder on the output encodes the same program, so the encoder with + // the next lower bitrate is this rendition's fallback. That track is published + // first, so this one can sit one priority below it and the allocator funds it first. + obs_encoder_t *lower = nullptr; + int lower_kbps = 0; + for (uint32_t idx = 0; idx < MAX_OUTPUT_VIDEO_ENCODERS; idx++) { + obs_encoder_t *other = obs_output_get_video_encoder2(output, idx); + if (!other || other == encoder) + continue; + // An encoder already publishing may have been retuned, so read the bitrate it + // was configured with. + int kbps = 0; + if (auto known = video_tracks.find(other); known != video_tracks.end()) { + kbps = (int)(known->second.bitrate_bps / 1000); + } else { + OBSDataAutoRelease other_settings = obs_encoder_get_settings(other); + kbps = other_settings ? (int)obs_data_get_int(other_settings, "bitrate") : 0; + } + if (kbps > lower_kbps && kbps < video_bitrate_kbps) { + lower = other; + lower_kbps = kbps; + } + } + std::optional lower_bps; + if (lower) { + if (video_tracks.find(lower) == video_tracks.end()) { + // The lower track keeps the headers it starts with, and some encoders only have + // them after their first frame. Wait for its first packet to start it instead; + // this rendition opens on its next keyframe after that. + uint8_t *lower_data = nullptr; + size_t lower_size = 0; + if (!obs_encoder_get_extra_data(lower, &lower_data, &lower_size) || lower_size == 0) + return; + VideoInit(lower); + } + // The lower rendition waits on one further down, so this one waits with it. + auto below = video_tracks.find(lower); + if (below == video_tracks.end()) + return; + if (below->second.producer) { + config.lower = below->second.producer; + lower_bps = (uint64_t)lower_kbps * 1000ULL; + } + } + auto track = moq::MediaTrackProducer::video(broadcast, moq::MediaTarget::kNamed{}, config); if (!track) { LOG_ERROR("Failed to initialize video track: %s", track.error().to_string().c_str()); @@ -772,7 +870,8 @@ void MoQOutput::VideoInit(obs_encoder_t *encoder) // bitrate drives; a quality target (CQP, CRF) only reserves. const bool dynamic = obs_encoder_get_caps(encoder) & OBS_ENCODER_CAP_DYN_BITRATE; if (dynamic && BitrateDriven(rate_control)) { - entry.rate.emplace(entry.bitrate_bps); + entry.rate = moq::RateControl::init(moq::RatePolicy{entry.bitrate_bps, lower_bps}); + entry.applied_bps = entry.bitrate_bps; entry.encoder = OBSGetWeakRef(encoder); } else { LOG_INFO("Video bitrate stays at %d kbps: the encoder cannot follow the bandwidth grant (%s)", @@ -781,7 +880,11 @@ void MoQOutput::VideoInit(obs_encoder_t *encoder) } video_tracks[encoder] = std::move(entry); - LOG_INFO("Video track initialized (%ux%u, %d kbps)", video_width, video_height, video_bitrate_kbps); + if (lower_bps) + LOG_INFO("Video track initialized (%ux%u, %d kbps, above a %d kbps rendition)", video_width, + video_height, video_bitrate_kbps, lower_kbps); + else + LOG_INFO("Video track initialized (%ux%u, %d kbps)", video_width, video_height, video_bitrate_kbps); } // NOTE: Caller must hold media_mutex. diff --git a/cpp/obs/src/moq-output.h b/cpp/obs/src/moq-output.h index f77ac44e0a3..796c5e516e1 100644 --- a/cpp/obs/src/moq-output.h +++ b/cpp/obs/src/moq-output.h @@ -15,7 +15,6 @@ #include #include #include "logger.h" -#include "moq-rate.h" #include "moq-worker.h" class MoQOutput { @@ -65,8 +64,9 @@ class MoQOutput { void CopyLastFailure(int *code, std::string *reason); // The bitrate the video encoders follow the bandwidth grant at, summed over - // every encoder that follows, beside the configured rate that caps it. - // False when no video encoder follows the grant. + // every encoder that follows, beside the configured rate that caps it. A rung + // disabled for lack of bandwidth counts as zero. False when no video encoder + // follows the grant. struct VideoRate { uint64_t target_bps = 0; uint64_t configured_bps = 0; @@ -126,10 +126,20 @@ class MoQOutput { uint64_t bitrate_bps = 0; // Claimed on the first frame after the session connects. std::shared_ptr reservation; - // Video only, when the encoder can retune: the grant it follows, and a + // Video only, when the encoder can retune: the grant it follows, which + // disables it while the grant cannot fund it above a lower rendition, and a // weak ref so Reset() can hand the configured bitrate back. - std::optional rate; + std::shared_ptr rate; OBSWeakEncoder encoder; + // The bitrate the encoder was last told, which Reset() restores when it moved. + uint64_t applied_bps = 0; + // Whether the last target enabled this rendition. + bool enabled = true; + // False from a disable until the keyframe that resumes it, dropping packets. + bool sending = true; + // Whether a frame was written. Until one is, the importer still holds the + // init's parameter sets, which a discontinuity would discard. + bool started = false; }; void VideoInit(obs_encoder_t *encoder); @@ -138,6 +148,9 @@ class MoQOutput { void AudioData(struct encoder_packet *packet); // NOTE: Caller must hold media_mutex. void Reserve(Track &track); + // Apply a rate target to `encoder`'s track: retune it, or stop sending it. + // NOTE: Caller must hold media_mutex. + void Retarget(obs_encoder_t *encoder, Track &track, const moq::RateTarget &target); obs_output_t *output; diff --git a/cpp/obs/src/moq-rate.h b/cpp/obs/src/moq-rate.h deleted file mode 100644 index 4b30ed207fe..00000000000 --- a/cpp/obs/src/moq-rate.h +++ /dev/null @@ -1,62 +0,0 @@ -// SPDX-License-Identifier: GPL-2.0-or-later -#pragma once - -#include -#include -#include -#include - -// Turns a reservation's grant into the bitrate an OBS encoder should run at, -// with the same shape as moq_mux::rate::Control: drops apply at once, raises -// climb at most 25% per second, and moves under 5% are ignored so a jittering -// grant doesn't retune the encoder on every frame. Keep the two in step. -class MoQRate { -public: - using Clock = std::chrono::steady_clock; - - // Starts at `max`, the configured bitrate, which the grant can only lower. - // The floor is a tenth of it: below that the video isn't worth sending. - explicit MoQRate(uint64_t max) : max(max), min(max / 10), target(max) {} - - uint64_t Target() const { return target; } - uint64_t Max() const { return max; } - - // Feed a grant, returning the new target when it moved enough to apply. - // A missing grant (no estimate) holds the target: losing the estimate is not - // evidence the uplink got better. - std::optional Update(std::optional grant, Clock::time_point now) - { - if (!grant) - return std::nullopt; - - const uint64_t desired = std::clamp(*grant, min, max); - uint64_t next = desired; - if (desired > target && applied) { - const double elapsed = std::chrono::duration(now - *applied).count(); - const auto grown = static_cast(static_cast(target) * - (1.0 + RAMP * std::max(elapsed, 0.0))); - next = std::clamp(std::min(grown, desired), min, max); - } - - // Hysteresis leaves `applied` alone, so suppressed raises keep accruing - // ramp allowance. The final step onto `max` is exempt, or a recovery - // would stall a few percent under the configured rate for good. - const bool settling = next > target && next == max; - const uint64_t diff = next > target ? next - target : target - next; - if (!settling && static_cast(diff) < static_cast(target) * HYSTERESIS) - return std::nullopt; - - target = next; - applied = now; - return next; - } - -private: - static constexpr double HYSTERESIS = 0.05; - static constexpr double RAMP = 0.25; - - uint64_t max; - uint64_t min; - uint64_t target; - std::optional applied; -}; diff --git a/cpp/obs/test/moq-output-test.cpp b/cpp/obs/test/moq-output-test.cpp index 975f97b7cc9..4c8ae88281a 100644 --- a/cpp/obs/test/moq-output-test.cpp +++ b/cpp/obs/test/moq-output-test.cpp @@ -9,6 +9,7 @@ // // Run with `just obs test` (ThreadSanitizer) or `just obs ci` (plain). This is not // part of the plugin build. +#include #include #include #include @@ -77,6 +78,14 @@ const std::vector g_opus_head = TestOpusHead(); obs_encoder_t *const VIDEO_ENCODER = reinterpret_cast(0x2); obs_encoder_t *const AUDIO_ENCODER = reinterpret_cast(0x3); +// A second, 2 Mbps video encoder below VIDEO_ENCODER's 8 Mbps, while g_ladder is set. +obs_encoder_t *const LOW_ENCODER = reinterpret_cast(0x5); +std::atomic g_ladder{false}; +// A third, 4 Mbps video encoder between the two, while g_ladder and g_mid are set. +obs_encoder_t *const MID_ENCODER = reinterpret_cast(0x6); +std::atomic g_mid{false}; +// LOW_ENCODER has no headers yet, as an encoder that only has them after its first frame. +std::atomic g_low_headers_pending{false}; } // namespace extern "C" { @@ -114,7 +123,11 @@ void obs_data_release(obs_data_t *) {} obs_encoder_t *obs_output_get_video_encoder2(const obs_output_t *, size_t idx) { - return idx == 0 ? VIDEO_ENCODER : nullptr; + if (idx == 0) + return VIDEO_ENCODER; + if (idx == 1 && g_ladder) + return LOW_ENCODER; + return idx == 2 && g_ladder && g_mid ? MID_ENCODER : nullptr; } bool obs_output_begin_data_capture(obs_output_t *, uint32_t) @@ -151,6 +164,11 @@ void obs_register_output_s(const struct obs_output_info *, size_t) {} bool obs_encoder_get_extra_data(const obs_encoder_t *encoder, uint8_t **data, size_t *size) { + if (encoder == LOW_ENCODER && g_low_headers_pending) { + *data = nullptr; + *size = 0; + return false; + } const auto &extra = encoder == AUDIO_ENCODER ? g_opus_head : g_h264_init; *data = const_cast(extra.data()); *size = extra.size(); @@ -163,10 +181,11 @@ const char *obs_encoder_get_codec(const obs_encoder_t *encoder) } // VideoInit reads coded size and CBR from the encoder. Any new libobs call in -// src/moq-output.cpp needs a stub here or `just obs ci` fails at link. -obs_data_t *obs_encoder_get_settings(const obs_encoder_t *) +// src/moq-output.cpp needs a stub here or `just obs ci` fails at link. The +// settings are the encoder itself, so each reports its own bitrate. +obs_data_t *obs_encoder_get_settings(const obs_encoder_t *encoder) { - return reinterpret_cast(0x4); + return reinterpret_cast(const_cast(encoder)); } uint32_t obs_encoder_get_width(const obs_encoder_t *) @@ -184,9 +203,11 @@ const char *obs_data_get_string(obs_data_t *, const char *) return g_rate_control.c_str(); } -long long obs_data_get_int(obs_data_t *, const char *) +long long obs_data_get_int(obs_data_t *settings, const char *) { - return 8000; + if (settings == reinterpret_cast(LOW_ENCODER)) + return 2000; + return settings == reinterpret_cast(MID_ENCODER) ? 4000 : 8000; } // The settings a retune builds carry only the bitrate, so the stub is that bitrate. @@ -298,6 +319,9 @@ void reset(std::string url) g_settings_error.clear(); g_rate_control = "CBR"; g_encoder_caps = OBS_ENCODER_CAP_DYN_BITRATE; + g_ladder = false; + g_mid = false; + g_low_headers_pending = false; { std::lock_guard lock(g_updates_mutex); g_updates.clear(); @@ -344,8 +368,12 @@ std::shared_ptr relayBroadcast(TestRelay &relay) return TestOk(available.get(), "available"); } -// The broadcast's catalog, once it has video and audio. -std::optional relayCatalog(const std::shared_ptr &broadcast) +// The broadcast's first catalog `ready` accepts, by default one with video and audio. +std::optional relayCatalog( + const std::shared_ptr &broadcast, + std::function ready = [](const moq::Catalog &catalog) { + return !catalog.video.empty() && !catalog.audio.empty(); + }) { if (!broadcast) return std::nullopt; @@ -357,7 +385,7 @@ std::optional relayCatalog(const std::shared_ptrvideo.empty() && !catalog->audio.empty()) + if (ready(*catalog)) return catalog; } } @@ -435,14 +463,38 @@ struct MoQOutputStatsTest { } return true; } - // Drive every following video track to its floor, as a collapsed uplink would. - static void Collapse(MoQOutput &output) + // Open `encoder`'s track from its extra data, as its first keyframe does just + // before the track's rate control is first read. + static void Init(MoQOutput &output, obs_encoder_t *encoder) + { + std::lock_guard lock(output.media_mutex); + output.VideoInit(encoder); + } + // Apply `target` to `encoder`'s track, as a grant the track's rate control returned. + static void Retarget(MoQOutput &output, obs_encoder_t *encoder, const moq::RateTarget &target) { std::lock_guard lock(output.media_mutex); - for (auto &[encoder, track] : output.video_tracks) { - if (track.rate) - track.rate->Update(1, MoQRate::Clock::now()); + auto it = output.video_tracks.find(encoder); + if (it != output.video_tracks.end()) + output.Retarget(encoder, it->second, target); + } + // The name and publisher priority of `encoder`'s track, or nothing without one. + static std::optional> Published(MoQOutput &output, obs_encoder_t *encoder) + { + std::shared_ptr broadcast; + std::shared_ptr producer; + { + std::lock_guard lock(output.media_mutex); + auto it = output.video_tracks.find(encoder); + if (it == output.video_tracks.end() || !it->second.producer) + return std::nullopt; + broadcast = output.broadcast; + producer = it->second.producer; } + auto name = TestOk(producer->demand(), "demand")->name(); + auto consumer = TestOk(broadcast->consume(), "consume"); + auto track = TestOk(consumer->subscribe_track(name, std::nullopt).get(), "subscribe_track"); + return std::make_pair(name, track->info().priority); } }; @@ -617,7 +669,8 @@ int main() auto keyframe = TestH264Keyframe(); auto video = videoPacket(keyframe, 0); o.Data(&video); - MoQOutputStatsTest::Collapse(o); + // The floor a collapsed uplink drives it to. + MoQOutputStatsTest::Retarget(o, VIDEO_ENCODER, moq::RateTarget::kEnabled{800'000}); MoQOutput::VideoRate rate; CHECK(o.TryGetVideoRate(&rate)); CHECK(rate.target_bps == 800'000); @@ -628,6 +681,164 @@ int main() } printf("stop restores the configured bitrate: ok\n"); + // Two video encoders form a ladder: the higher rung sits one priority below the + // lower, and a disabled rung leaves the catalog disabled, dropping its packets + // until a keyframe after the grant funds it again. No encoder follows here, so + // only the targets the test applies move it. + { + TestRelay relay; + reset(relay.Url()); + g_ladder = true; + g_encoder_caps = 0; + MoQOutput o(nullptr, OUTPUT); + CHECK(o.Start()); + CHECK(WaitFor([&] { return o.IsLiveSession(); })); + + auto keyframe = TestH264Keyframe(); + std::vector delta = {0x00, 0x00, 0x00, 0x01, 0x41, 0x9a, 0x00, 0x33}; + auto video = videoPacket(keyframe, 0); + o.Data(&video); + size_t sent = keyframe.size(); + CHECK(o.GetTotalBytes() == sent); + + const auto high = MoQOutputStatsTest::Published(o, VIDEO_ENCODER); + const auto low = MoQOutputStatsTest::Published(o, LOW_ENCODER); + CHECK(high.has_value() && low.has_value()); + if (high && low) { + CHECK(high->second == low->second - 1); + auto broadcast = relayBroadcast(relay); + + MoQOutputStatsTest::Retarget(o, VIDEO_ENCODER, moq::RateTarget::kDisabled{}); + auto disabled = relayCatalog(broadcast, [&](const moq::Catalog &catalog) { + auto it = catalog.video.find(high->first); + return it != catalog.video.end() && !it->second.enabled; + }); + CHECK(disabled.has_value()); + if (disabled) { + auto lower = disabled->video.find(low->first); + CHECK(lower != disabled->video.end() && lower->second.enabled); + } + + auto dropped = videoPacket(delta, 1); + dropped.keyframe = false; + o.Data(&dropped); + auto skipped = videoPacket(keyframe, 2); + o.Data(&skipped); + CHECK(o.GetTotalBytes() == sent); + + // Funded again, it still waits for a keyframe to resume on. + MoQOutputStatsTest::Retarget(o, VIDEO_ENCODER, moq::RateTarget::kEnabled{4'000'000}); + auto early = videoPacket(delta, 3); + early.keyframe = false; + o.Data(&early); + CHECK(o.GetTotalBytes() == sent); + auto resumed = videoPacket(keyframe, 4); + o.Data(&resumed); + sent += keyframe.size(); + CHECK(o.GetTotalBytes() == sent); + auto enabled = relayCatalog(broadcast, [&](const moq::Catalog &catalog) { + auto it = catalog.video.find(high->first); + return it != catalog.video.end() && it->second.enabled; + }); + CHECK(enabled.has_value()); + } + o.Stop(); + } + printf("a ladder rung disables, drops packets, and resumes on a keyframe: ok\n"); + + // A rung disabled before its first frame keeps the extra data's parameter sets, + // so a keyframe without them still resumes it decodable. + { + TestRelay relay; + reset(relay.Url()); + g_encoder_caps = 0; + MoQOutput o(nullptr, OUTPUT); + CHECK(o.Start()); + CHECK(WaitFor([&] { return o.IsLiveSession(); })); + + MoQOutputStatsTest::Init(o, VIDEO_ENCODER); + MoQOutputStatsTest::Retarget(o, VIDEO_ENCODER, moq::RateTarget::kDisabled{}); + MoQOutputStatsTest::Retarget(o, VIDEO_ENCODER, moq::RateTarget::kEnabled{4'000'000}); + std::vector idr = {0x00, 0x00, 0x00, 0x01, 0x65, 0x88, 0x84, 0x00, 0x33, 0xff}; + auto resumed = videoPacket(idr, 0); + o.Data(&resumed); + CHECK(o.GetTotalBytes() == idr.size()); + + auto broadcast = relayBroadcast(relay); + auto catalog = + relayCatalog(broadcast, [](const moq::Catalog &catalog) { return !catalog.video.empty(); }); + CHECK(catalog.has_value() && catalog->video.size() == 1); + if (catalog && catalog->video.size() == 1) { + const auto &[name, rendition] = *catalog->video.begin(); + auto media = TestOk( + moq::MediaTrackConsumer::subscribe(broadcast, {moq::Rendition::kVideo{name, rendition}}) + .get(), + "container consumer"); + auto next = media->next(); + CHECK(next.wait_for(std::chrono::seconds(10)) == std::future_status::ready); + auto first = TestOk(next.get(), "first frame"); + const auto *frame = first ? std::get_if(&first->get_variant()) + : nullptr; + CHECK(frame != nullptr); + if (frame) { + const auto &payload = frame->frame.payload; + // The start of the extra data's SPS, past its start code. + const auto init = TestH264Init(); + CHECK(std::search(payload.begin(), payload.end(), init.begin() + 4, + init.begin() + 12) != payload.end()); + } + } + o.Stop(); + } + printf("a rung disabled before its first frame keeps its parameter sets: ok\n"); + + // A rung whose lower encoder has no headers yet waits for the lower's first packet + // rather than start the lower track without them, and so does every rung above it. + // Each then opens on its next keyframe one priority below its lower. + { + TestRelay relay; + reset(relay.Url()); + g_ladder = true; + g_mid = true; + g_encoder_caps = 0; + g_low_headers_pending = true; + MoQOutput o(nullptr, OUTPUT); + CHECK(o.Start()); + CHECK(WaitFor([&] { return o.IsLiveSession(); })); + + auto keyframe = TestH264Keyframe(); + std::vector delta = {0x00, 0x00, 0x00, 0x01, 0x41, 0x9a, 0x00, 0x33}; + auto early = videoPacket(keyframe, 0); + o.Data(&early); + CHECK(o.GetTotalBytes() == 0); + CHECK(!MoQOutputStatsTest::Published(o, LOW_ENCODER).has_value()); + CHECK(!MoQOutputStatsTest::Published(o, MID_ENCODER).has_value()); + + g_low_headers_pending = false; + auto first = videoPacket(keyframe, 0); + first.encoder = LOW_ENCODER; + o.Data(&first); + size_t sent = keyframe.size(); + CHECK(o.GetTotalBytes() == sent); + + auto skipped = videoPacket(delta, 1); + skipped.keyframe = false; + o.Data(&skipped); + CHECK(o.GetTotalBytes() == sent); + auto opened = videoPacket(keyframe, 2); + o.Data(&opened); + sent += keyframe.size(); + CHECK(o.GetTotalBytes() == sent); + + const auto high = MoQOutputStatsTest::Published(o, VIDEO_ENCODER); + const auto mid = MoQOutputStatsTest::Published(o, MID_ENCODER); + const auto low = MoQOutputStatsTest::Published(o, LOW_ENCODER); + CHECK(high.has_value() && mid.has_value() && low.has_value()); + CHECK(high && mid && low && high->second == mid->second - 1 && mid->second == low->second - 1); + o.Stop(); + } + printf("a ladder rung waits for its lower rung's headers: ok\n"); + // Stop drains the session, even when the output is destroyed straight after: a // large frame written just before it still reaches the relay, and the track then // ends cleanly instead of being cut off with the session. diff --git a/cpp/obs/test/moq-rate-test.cpp b/cpp/obs/test/moq-rate-test.cpp deleted file mode 100644 index c14155f18be..00000000000 --- a/cpp/obs/test/moq-rate-test.cpp +++ /dev/null @@ -1,87 +0,0 @@ -// SPDX-License-Identifier: GPL-2.0-or-later -// -// The cases moq_mux::rate::Control is tested with, so the two policies agree. -#include "moq-rate.h" - -#include -#include - -using namespace std::chrono_literals; - -namespace { -int g_failures = 0; - -#define CHECK(cond) \ - do { \ - if (!(cond)) { \ - fprintf(stderr, "FAIL %s:%d: %s\n", __FILE__, __LINE__, #cond); \ - g_failures++; \ - } \ - } while (0) - -const MoQRate::Clock::time_point START{}; -} // namespace - -int main() -{ - { - // Starts at the configured rate and drops at once. - MoQRate rate(4'000'000); - CHECK(rate.Target() == 4'000'000); - CHECK(rate.Update(2'000'000, START) == 2'000'000); - } - { - // A missing grant holds, however long it lasts. - MoQRate rate(4'000'000); - rate.Update(2'000'000, START); - CHECK(!rate.Update(std::nullopt, START + 10s)); - CHECK(rate.Target() == 2'000'000); - } - { - // Never above the configured rate, never below a tenth of it. - MoQRate rate(4'000'000); - CHECK(!rate.Update(100'000'000, START)); - CHECK(rate.Update(1, START) == 400'000); - } - { - // One second of ramp allows 1M -> 1.25M, not the whole grant. - MoQRate rate(4'000'000); - rate.Update(1'000'000, START); - CHECK(rate.Update(4'000'000, START + 1s) == 1'250'000); - } - { - // Each 100ms tick alone is under the 5% deadband; suppression must keep - // the ramp anchor so the raise lands on the second tick. - MoQRate rate(4'000'000); - rate.Update(1'000'000, START); - CHECK(!rate.Update(4'000'000, START + 100ms)); - CHECK(rate.Update(4'000'000, START + 200ms) == 1'050'000); - } - { - // A recovery lands exactly on the configured rate, then stops retuning. - MoQRate rate(4'000'000); - rate.Update(1'000'000, START); - std::optional last; - for (int tick = 1; tick <= 200; tick++) { - if (auto next = rate.Update(4'000'000, START + tick * 100ms)) - last = next; - } - CHECK(last == 4'000'000); - CHECK(!rate.Update(4'000'000, START + 60s)); - } - { - // Small moves either way stay inside the deadband. - MoQRate rate(4'000'000); - rate.Update(2'000'000, START); - CHECK(!rate.Update(2'010'000, START + 100ms)); - CHECK(!rate.Update(1'960'000, START + 1s)); - CHECK(rate.Update(1'600'000, START + 2s) == 1'600'000); - } - - if (g_failures) { - fprintf(stderr, "%d rate check(s) failed\n", g_failures); - return EXIT_FAILURE; - } - std::puts("rate follows the grant like moq_mux::rate::Control: ok"); - return EXIT_SUCCESS; -} diff --git a/dart/moq/lib/src/aliases.dart b/dart/moq/lib/src/aliases.dart index 6d446d10505..a9c38dde51b 100644 --- a/dart/moq/lib/src/aliases.dart +++ b/dart/moq/lib/src/aliases.dart @@ -170,6 +170,15 @@ typedef BandwidthConsumer = MoqBandwidthConsumer; /// One track's standing claim on an [Allocator]. typedef Reservation = MoqReservation; +/// Turns a [Reservation]'s grant into a sender's [RateTarget]; call `update` on every frame. +typedef RateControl = MoqRateControl; + +/// How a [RateControl] maps a grant onto a bitrate, and the lower rendition that disables it. +typedef RatePolicy = MoqRatePolicy; + +/// What a sender following a [RateControl] should be doing: an enabled bitrate, or disabled. +typedef RateTarget = MoqRateTarget; + /// A snapshot of transport connection statistics. typedef ConnectionStats = MoqConnectionStats; diff --git a/dart/moq_ffi/lib/src/moq.dart b/dart/moq_ffi/lib/src/moq.dart index f47dc76ce53..a0c2169c173 100644 --- a/dart/moq_ffi/lib/src/moq.dart +++ b/dart/moq_ffi/lib/src/moq.dart @@ -648,6 +648,65 @@ class FfiConverterMoqGrant { } } +class MoqRatePolicy { + final int maxBps; + final int? lowerBps; + MoqRatePolicy({required this.maxBps, this.lowerBps = null}); +} + +class FfiConverterMoqRatePolicy { + static MoqRatePolicy lift(RustBuffer buf) { + return FfiConverterMoqRatePolicy.read(buf.asUint8List()).value; + } + + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final maxBps_lifted = FfiConverterUInt64.read( + Uint8List.view(buf.buffer, new_offset), + ); + final maxBps = maxBps_lifted.value; + new_offset += maxBps_lifted.bytesRead; + final lowerBps_lifted = FfiConverterOptionalUInt64.read( + Uint8List.view(buf.buffer, new_offset), + ); + final lowerBps = lowerBps_lifted.value; + new_offset += lowerBps_lifted.bytesRead; + return LiftRetVal( + MoqRatePolicy(maxBps: maxBps, lowerBps: lowerBps), + new_offset - buf.offsetInBytes, + ); + } + + static RustBuffer lower(MoqRatePolicy value) { + final total_length = + FfiConverterUInt64.allocationSize(value.maxBps) + + FfiConverterOptionalUInt64.allocationSize(value.lowerBps) + + 0; + final buf = Uint8List(total_length); + write(value, buf); + return toRustBuffer(buf); + } + + static int write(MoqRatePolicy value, Uint8List buf) { + int new_offset = buf.offsetInBytes; + new_offset += FfiConverterUInt64.write( + value.maxBps, + Uint8List.view(buf.buffer, new_offset), + ); + new_offset += FfiConverterOptionalUInt64.write( + value.lowerBps, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset - buf.offsetInBytes; + } + + static int allocationSize(MoqRatePolicy value) { + return FfiConverterUInt64.allocationSize(value.maxBps) + + FfiConverterOptionalUInt64.allocationSize(value.lowerBps) + + 0; + } +} + class MoqFetchGroupOptions { final int priority; final Duration? maxDelay; @@ -2487,11 +2546,13 @@ class MoqVideoInit { final Uint8List data; final String? label; final MoqVideoHint? hint; + final MoqMediaTrackProducer? lower; MoqVideoInit({ required this.format, required this.data, this.label = null, this.hint = null, + this.lower = null, }); } @@ -2522,8 +2583,19 @@ class FfiConverterMoqVideoInit { ); final hint = hint_lifted.value; new_offset += hint_lifted.bytesRead; + final lower_lifted = FfiConverterOptionalMoqMediaTrackProducer.read( + Uint8List.view(buf.buffer, new_offset), + ); + final lower = lower_lifted.value; + new_offset += lower_lifted.bytesRead; return LiftRetVal( - MoqVideoInit(format: format, data: data, label: label, hint: hint), + MoqVideoInit( + format: format, + data: data, + label: label, + hint: hint, + lower: lower, + ), new_offset - buf.offsetInBytes, ); } @@ -2534,6 +2606,7 @@ class FfiConverterMoqVideoInit { FfiConverterUint8List.allocationSize(value.data) + FfiConverterOptionalString.allocationSize(value.label) + FfiConverterOptionalMoqVideoHint.allocationSize(value.hint) + + FfiConverterOptionalMoqMediaTrackProducer.allocationSize(value.lower) + 0; final buf = Uint8List(total_length); write(value, buf); @@ -2558,6 +2631,10 @@ class FfiConverterMoqVideoInit { value.hint, Uint8List.view(buf.buffer, new_offset), ); + new_offset += FfiConverterOptionalMoqMediaTrackProducer.write( + value.lower, + Uint8List.view(buf.buffer, new_offset), + ); return new_offset - buf.offsetInBytes; } @@ -2566,6 +2643,7 @@ class FfiConverterMoqVideoInit { FfiConverterUint8List.allocationSize(value.data) + FfiConverterOptionalString.allocationSize(value.label) + FfiConverterOptionalMoqVideoHint.allocationSize(value.hint) + + FfiConverterOptionalMoqMediaTrackProducer.allocationSize(value.lower) + 0; } } @@ -5144,6 +5222,120 @@ class FfiConverterMoqAudioSampleFormat { } } +abstract class MoqRateTarget { + RustBuffer lower(); + int allocationSize(); + int write(Uint8List buf); +} + +class FfiConverterMoqRateTarget { + static MoqRateTarget lift(RustBuffer buffer) { + return FfiConverterMoqRateTarget.read(buffer.asUint8List()).value; + } + + static LiftRetVal read(Uint8List buf) { + final index = buf.buffer.asByteData(buf.offsetInBytes).getInt32(0); + final subview = Uint8List.view(buf.buffer, buf.offsetInBytes + 4); + switch (index) { + case 1: + final lifted = EnabledMoqRateTarget.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); + case 2: + final lifted = DisabledMoqRateTarget.read(subview); + return LiftRetVal( + lifted.value, + lifted.bytesRead - subview.offsetInBytes + 4, + ); + default: + throw UniffiInternalError( + UniffiInternalError.unexpectedEnumCase, + "Unable to determine enum variant", + ); + } + } + + static RustBuffer lower(MoqRateTarget value) { + return value.lower(); + } + + static int allocationSize(MoqRateTarget value) { + return value.allocationSize(); + } + + static int write(MoqRateTarget value, Uint8List buf) { + return value.write(buf) - buf.offsetInBytes; + } +} + +class EnabledMoqRateTarget extends MoqRateTarget { + final int bitrateBps; + EnabledMoqRateTarget(int this.bitrateBps); + EnabledMoqRateTarget._(int this.bitrateBps); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + final bitrateBps_lifted = FfiConverterUInt64.read( + Uint8List.view(buf.buffer, new_offset), + ); + final bitrateBps = bitrateBps_lifted.value; + new_offset += bitrateBps_lifted.bytesRead; + return LiftRetVal(EnabledMoqRateTarget._(bitrateBps), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return FfiConverterUInt64.allocationSize(bitrateBps) + 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 1); + int new_offset = buf.offsetInBytes + 4; + new_offset += FfiConverterUInt64.write( + bitrateBps, + Uint8List.view(buf.buffer, new_offset), + ); + return new_offset; + } +} + +class DisabledMoqRateTarget extends MoqRateTarget { + DisabledMoqRateTarget(); + DisabledMoqRateTarget._(); + static LiftRetVal read(Uint8List buf) { + int new_offset = buf.offsetInBytes; + return LiftRetVal(DisabledMoqRateTarget._(), new_offset); + } + + @override + RustBuffer lower() { + final buf = Uint8List(allocationSize()); + write(buf); + return toRustBuffer(buf); + } + + @override + int allocationSize() { + return 4; + } + + @override + int write(Uint8List buf) { + buf.buffer.asByteData(buf.offsetInBytes).setInt32(0, 2); + int new_offset = buf.offsetInBytes + 4; + return new_offset; + } +} + abstract class MoqAnchor { RustBuffer lower(); int allocationSize(); @@ -9653,6 +9845,108 @@ class FfiConverterMoqBandwidthConsumer { } } +abstract class MoqRateControlInterface { + MoqRateTarget target(); + MoqRateTarget? update({required MoqReservation reservation}); +} + +final _MoqRateControlFinalizer = Finalizer>((ptr) { + rustCall((status) => uniffi_moq_ffi_fn_free_moqratecontrol(ptr, status)); +}); + +class MoqRateControl implements MoqRateControlInterface { + late final Pointer _ptr; + bool _isDisposed = false; + MoqRateControl._(this._ptr) { + _MoqRateControlFinalizer.attach(this, _ptr, detach: this); + } + MoqRateControl({required MoqRatePolicy policy}) + : _ptr = rustCall( + (status) => uniffi_moq_ffi_fn_constructor_moqratecontrol_new( + FfiConverterMoqRatePolicy.lower(policy), + status, + ), + null, + ) { + _MoqRateControlFinalizer.attach(this, _ptr, detach: this); + } + factory MoqRateControl.lift(Pointer ptr) { + return MoqRateControl._(ptr); + } + Pointer uniffiClonePointer() { + if (_isDisposed) { + throw StateError("Object has already been disposed"); + } + return rustCall( + (status) => uniffi_moq_ffi_fn_clone_moqratecontrol(_ptr, status), + ); + } + + void dispose() { + if (_isDisposed) { + return; + } + final status = calloc(); + _isDisposed = true; + _MoqRateControlFinalizer.detach(this); + try { + uniffi_moq_ffi_fn_free_moqratecontrol(_ptr, status); + checkCallStatus(NullRustCallStatusErrorHandler(), status); + } finally { + calloc.free(status); + } + } + + MoqRateTarget target() { + return rustCallWithLifter( + (status) => uniffi_moq_ffi_fn_method_moqratecontrol_target( + uniffiClonePointer(), + status, + ), + FfiConverterMoqRateTarget.lift, + null, + ); + } + + MoqRateTarget? update({required MoqReservation reservation}) { + return rustCallWithLifter( + (status) => uniffi_moq_ffi_fn_method_moqratecontrol_update( + uniffiClonePointer(), + FfiConverterMoqReservation.lower(reservation), + status, + ), + FfiConverterOptionalMoqRateTarget.lift, + null, + ); + } +} + +class FfiConverterMoqRateControl { + static MoqRateControl lift(Pointer ptr) { + return MoqRateControl.lift(ptr); + } + + static Pointer lower(MoqRateControl value) { + return value.uniffiClonePointer(); + } + + static int allocationSize(MoqRateControl value) { + return 8; + } + + static LiftRetVal read(Uint8List buf) { + final handle = buf.buffer.asByteData(buf.offsetInBytes).getInt64(0); + final pointer = Pointer.fromAddress(handle); + return LiftRetVal(MoqRateControl.lift(pointer), 8); + } + + static int write(MoqRateControl value, Uint8List buf) { + final handle = lower(value); + buf.buffer.asByteData(buf.offsetInBytes).setInt64(0, handle.address); + return 8; + } +} + abstract class MoqReservationInterface { int? rate(); void update({required int maxBps}); @@ -11673,6 +11967,7 @@ abstract class MoqMediaTrackProducerInterface { void discontinuity(); void finish(); void seek({required int sequence}); + void setEnabled({required bool enabled}); void writeFrame({required MoqFrame frame}); } @@ -11793,6 +12088,16 @@ class MoqMediaTrackProducer implements MoqMediaTrackProducerInterface { }, moqExceptionErrorHandler); } + void setEnabled({required bool enabled}) { + return rustCall((status) { + uniffi_moq_ffi_fn_method_moqmediatrackproducer_set_enabled( + uniffiClonePointer(), + FfiConverterBool.lower(enabled), + status, + ); + }, moqExceptionErrorHandler); + } + void writeFrame({required MoqFrame frame}) { return rustCall((status) { uniffi_moq_ffi_fn_method_moqmediatrackproducer_write_frame( @@ -16402,6 +16707,58 @@ class FfiConverterOptionalMoqMediaTimebase { } } +class FfiConverterOptionalMoqMediaTrackProducer { + static MoqMediaTrackProducer? lift(RustBuffer buf) { + return FfiConverterOptionalMoqMediaTrackProducer.read( + buf.asUint8List(), + ).value; + } + + static LiftRetVal read(Uint8List buf) { + if (ByteData.view(buf.buffer, buf.offsetInBytes).getInt8(0) == 0) { + return LiftRetVal(null, 1); + } + final result = FfiConverterMoqMediaTrackProducer.read( + Uint8List.view(buf.buffer, buf.offsetInBytes + 1), + ); + return LiftRetVal( + result.value, + result.bytesRead + 1, + ); + } + + static int allocationSize([MoqMediaTrackProducer? value]) { + if (value == null) { + return 1; + } + return FfiConverterMoqMediaTrackProducer.allocationSize(value) + 1; + } + + static RustBuffer lower(MoqMediaTrackProducer? value) { + if (value == null) { + return toRustBuffer(Uint8List.fromList([0])); + } + final buf = Uint8List( + FfiConverterOptionalMoqMediaTrackProducer.allocationSize(value), + ); + FfiConverterOptionalMoqMediaTrackProducer.write(value, buf); + return toRustBuffer(buf); + } + + static int write(MoqMediaTrackProducer? value, Uint8List buf) { + if (value == null) { + buf[0] = 0; + return 1; + } + buf[0] = 1; + return FfiConverterMoqMediaTrackProducer.write( + value, + Uint8List.view(buf.buffer, buf.offsetInBytes + 1), + ) + + 1; + } +} + class FfiConverterOptionalMoqOrder { static MoqOrder? lift(RustBuffer buf) { return FfiConverterOptionalMoqOrder.read(buf.asUint8List()).value; @@ -16541,6 +16898,53 @@ class FfiConverterOptionalMoqPosition { } } +class FfiConverterOptionalMoqRateTarget { + static MoqRateTarget? lift(RustBuffer buf) { + return FfiConverterOptionalMoqRateTarget.read(buf.asUint8List()).value; + } + + static LiftRetVal read(Uint8List buf) { + if (ByteData.view(buf.buffer, buf.offsetInBytes).getInt8(0) == 0) { + return LiftRetVal(null, 1); + } + final result = FfiConverterMoqRateTarget.read( + Uint8List.view(buf.buffer, buf.offsetInBytes + 1), + ); + return LiftRetVal(result.value, result.bytesRead + 1); + } + + static int allocationSize([MoqRateTarget? value]) { + if (value == null) { + return 1; + } + return FfiConverterMoqRateTarget.allocationSize(value) + 1; + } + + static RustBuffer lower(MoqRateTarget? value) { + if (value == null) { + return toRustBuffer(Uint8List.fromList([0])); + } + final buf = Uint8List( + FfiConverterOptionalMoqRateTarget.allocationSize(value), + ); + FfiConverterOptionalMoqRateTarget.write(value, buf); + return toRustBuffer(buf); + } + + static int write(MoqRateTarget? value, Uint8List buf) { + if (value == null) { + buf[0] = 0; + return 1; + } + buf[0] = 1; + return FfiConverterMoqRateTarget.write( + value, + Uint8List.view(buf.buffer, buf.offsetInBytes + 1), + ) + + 1; + } +} + class FfiConverterOptionalMoqRequest { static MoqRequest? lift(RustBuffer buf) { return FfiConverterOptionalMoqRequest.read(buf.asUint8List()).value; @@ -17999,6 +18403,47 @@ external RustBuffer uniffi_moq_ffi_fn_method_moqbandwidthconsumer_rate( external Pointer uniffi_moq_ffi_fn_method_moqbandwidthconsumer_rate_changed(Pointer ptr); +@Native Function(Pointer, Pointer)>( + assetId: _uniffiAssetId, +) +external Pointer uniffi_moq_ffi_fn_clone_moqratecontrol( + Pointer handle, + Pointer uniffiStatus, +); + +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_free_moqratecontrol( + Pointer handle, + Pointer uniffiStatus, +); + +@Native Function(RustBuffer, Pointer)>( + assetId: _uniffiAssetId, +) +external Pointer uniffi_moq_ffi_fn_constructor_moqratecontrol_new( + RustBuffer policy, + Pointer uniffiStatus, +); + +@Native, Pointer)>( + assetId: _uniffiAssetId, +) +external RustBuffer uniffi_moq_ffi_fn_method_moqratecontrol_target( + Pointer ptr, + Pointer uniffiStatus, +); + +@Native< + RustBuffer Function(Pointer, Pointer, Pointer) +>(assetId: _uniffiAssetId) +external RustBuffer uniffi_moq_ffi_fn_method_moqratecontrol_update( + Pointer ptr, + Pointer reservation, + Pointer uniffiStatus, +); + @Native Function(Pointer, Pointer)>( assetId: _uniffiAssetId, ) @@ -18883,6 +19328,15 @@ external void uniffi_moq_ffi_fn_method_moqmediatrackproducer_seek( Pointer uniffiStatus, ); +@Native, Int8, Pointer)>( + assetId: _uniffiAssetId, +) +external void uniffi_moq_ffi_fn_method_moqmediatrackproducer_set_enabled( + Pointer ptr, + int enabled, + Pointer uniffiStatus, +); + @Native, RustBuffer, Pointer)>( assetId: _uniffiAssetId, ) @@ -20701,6 +21155,12 @@ external int uniffi_moq_ffi_checksum_method_moqbandwidthconsumer_rate(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqbandwidthconsumer_rate_changed(); +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqratecontrol_target(); + +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqratecontrol_update(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqreservation_rate(); @@ -20880,6 +21340,9 @@ external int uniffi_moq_ffi_checksum_method_moqmediatrackproducer_finish(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqmediatrackproducer_seek(); +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_method_moqmediatrackproducer_set_enabled(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_method_moqmediatrackproducer_write_frame(); @@ -21237,6 +21700,9 @@ external int uniffi_moq_ffi_checksum_constructor_moqaudioencoder_new(); @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_constructor_moqaudioproducer_new(); +@Native(assetId: _uniffiAssetId) +external int uniffi_moq_ffi_checksum_constructor_moqratecontrol_new(); + @Native(assetId: _uniffiAssetId) external int uniffi_moq_ffi_checksum_constructor_moqjsonsnapshotconsumer_new(); @@ -21400,6 +21866,12 @@ void _checkApiChecksums() { 38021) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_method_moqratecontrol_target() != 6016) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } + if (uniffi_moq_ffi_checksum_method_moqratecontrol_update() != 28669) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_method_moqreservation_rate() != 65114) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } @@ -21591,6 +22063,10 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_method_moqmediatrackproducer_seek() != 37244) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_method_moqmediatrackproducer_set_enabled() != + 1046) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_method_moqmediatrackproducer_write_frame() != 7616) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); @@ -21957,6 +22433,9 @@ void _checkApiChecksums() { if (uniffi_moq_ffi_checksum_constructor_moqaudioproducer_new() != 51486) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); } + if (uniffi_moq_ffi_checksum_constructor_moqratecontrol_new() != 63053) { + throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); + } if (uniffi_moq_ffi_checksum_constructor_moqjsonsnapshotconsumer_new() != 11929) { throw UniffiInternalError.panicked("UniFFI API checksum mismatch"); diff --git a/doc/bin/obs.md b/doc/bin/obs.md index f50e77d4f66..bb2bbe2128d 100644 --- a/doc/bin/obs.md +++ b/doc/bin/obs.md @@ -67,6 +67,14 @@ Saving Settings > Stream afterwards replaces the MoQ service. The dock's Go Live publishes one rendition. OBS's Enhanced Broadcasting (multitrack video) always streams through its RTMP output, so it cannot drive MoQ. +A rendition above a lower-bitrate one stops sending while its share of the +estimate cannot fund it, advertised as disabled so viewers switch down. Only an encoder +that follows the estimate, as above, stops; one that cannot retune while live, +or uses CQP or CRF, keeps sending. A stopped encoder keeps encoding, since OBS +cannot pause one encoder of a live output, and resumes at that encoder's next +keyframe once the estimate recovers. This needs the session to probe for +bandwidth, which takes moq-quic with BBR3. + ## Source quality and moq-transcode Without simulcast layers, OBS publishes **one** rendition. When a diff --git a/doc/lib/cpp/index.md b/doc/lib/cpp/index.md index de471a58826..c5a4339f6c1 100644 --- a/doc/lib/cpp/index.md +++ b/doc/lib/cpp/index.md @@ -149,7 +149,7 @@ ok(publication->announce({})); // each broadcast mints a fresh instance ok(broadcast->close()); // keep the producer alive while publishing, then close explicitly ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(false)` after a `discontinuity()`, and `set_enabled(true)` on the keyframe that resumes it. ## Futures diff --git a/doc/lib/dart/index.md b/doc/lib/dart/index.md index 2fafb238017..d01410bb6da 100644 --- a/doc/lib/dart/index.md +++ b/doc/lib/dart/index.md @@ -139,7 +139,7 @@ mine.close(); await moq.close(); ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. ## Things to know diff --git a/doc/lib/go/index.md b/doc/lib/go/index.md index ff64d45c78e..3e5a5679925 100644 --- a/doc/lib/go/index.md +++ b/doc/lib/go/index.md @@ -119,7 +119,7 @@ _ = video.Finish() broadcast.Close() // keep the producer reachable while publishing, then close explicitly ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `Lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `Lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `Lower` in its `media.VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `SetEnabled(false)` after a `Discontinuity()`, and `SetEnabled(true)` on the keyframe that resumes it. ## Things to know diff --git a/doc/lib/kt/index.md b/doc/lib/kt/index.md index d781b00f704..719e662c68e 100644 --- a/doc/lib/kt/index.md +++ b/doc/lib/kt/index.md @@ -94,7 +94,7 @@ Moq.connect("https://relay.example.com").use { moq -> } ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Producer options hold input, codec settings, track naming, and optional session bandwidth ownership; standalone encoders take only codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded video track takes `lower` in its `VideoInit` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. ## Things to know diff --git a/doc/lib/py/index.md b/doc/lib/py/index.md index c21b9a5d9c7..52d1044c992 100644 --- a/doc/lib/py/index.md +++ b/doc/lib/py/index.md @@ -109,7 +109,7 @@ async def main(): asyncio.run(main()) ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `media.TrackProducer.video` takes `lower` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `set_enabled(False)` after a `discontinuity()`, and `set_enabled(True)` on the keyframe that resumes it. ## Things to know diff --git a/doc/lib/swift/index.md b/doc/lib/swift/index.md index 05f6eeac60d..cebbed1cda6 100644 --- a/doc/lib/swift/index.md +++ b/doc/lib/swift/index.md @@ -95,7 +95,7 @@ session.shutdown() try await session.closed() ``` -Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. +Raw video encoders take an exact numerator/denominator frame rate, such as 30000/1001. Pass the publication track name to the producer, alongside its input and codec settings. For simulcast, give each rendition the next one down as `lower`: it stops encoding, advertised as disabled, while its bandwidth grant cannot fund it, and resumes once the grant recovers. A rung only disables while the session probes for bandwidth, which on native transports takes moq-quic with BBR3. A pre-encoded `Media.TrackProducer.video` takes `lower` too, which only places it one priority below; follow its reservation with `RateControl`, and while it reports disabled, drop the frames, call `setEnabled(false)` after a `discontinuity()`, and `setEnabled(true)` on the keyframe that resumes it. ## Things to know diff --git a/go/wrapper/MIGRATE.md b/go/wrapper/MIGRATE.md index f0f803f0dde..8f0b8385259 100644 --- a/go/wrapper/MIGRATE.md +++ b/go/wrapper/MIGRATE.md @@ -4,6 +4,10 @@ How to update your code for each breaking change, newest first. ## Unreleased +### `media.VideoInit` is a struct + +`media.VideoInit` is its own struct rather than an alias of the generated record, so its new `Lower` field takes a `*media.TrackProducer`. A composite literal with `Format`, `Data`, `Label`, and `Hint` compiles unchanged; build one that way where you converted a generated `MoqVideoInit`. + ### Announcement captures are required ([#7484](https://github.com/moq-dev/moq/pull/7484)) `Announce.Captures` is `[]string` instead of `*[]string`. Remove pointer checks and dereferences; an empty slice means the filter has no wildcards. Regenerate bindings and rebuild with the matching native library. diff --git a/go/wrapper/media/media.go b/go/wrapper/media/media.go index ef704bf3cfd..bbee93deab5 100644 --- a/go/wrapper/media/media.go +++ b/go/wrapper/media/media.go @@ -61,10 +61,31 @@ type ( // AudioInit describes an audio codec and its initialization bytes. AudioInit = ffi.MoqAudioInit - // VideoInit describes a video codec, initialization bytes, and hints. - VideoInit = ffi.MoqVideoInit ) +// VideoInit describes a video codec, initialization bytes, and hints. +type VideoInit struct { + // Format selects the video codec. + Format VideoFormat + // Data supplies codec init bytes (an avcC, an hvcC, ...); may be empty for a format that resolves in band. + Data []byte + // Label is a human-readable rendition name for a track picker. + Label *string + // Hint carries catalog fields the stream cannot reveal itself. + Hint *VideoHint + // Lower is the next lower rendition of the same source: this track then sits one + // priority below it, so a bandwidth allocator funds that one first. + Lower *TrackProducer +} + +func (init VideoInit) ffi() ffi.MoqVideoInit { + var lower **ffi.MoqMediaTrackProducer + if init.Lower != nil { + lower = &init.Lower.inner + } + return ffi.MoqVideoInit{Format: init.Format, Data: init.Data, Label: init.Label, Hint: init.Hint, Lower: lower} +} + // Timebase shares a source's timestamp offset without keeping its broadcast open. type Timebase struct{ inner *ffi.MoqMediaTimebase } @@ -186,7 +207,7 @@ func NewVideoTrackProducer(broadcast *moq.BroadcastProducer, target Target, init if err != nil { return nil, err } - inner, err := ffi.MoqMediaTrackProducerVideo(bridge.BroadcastProducer(broadcast), resolved, init) + inner, err := ffi.MoqMediaTrackProducerVideo(bridge.BroadcastProducer(broadcast), resolved, init.ffi()) if err != nil { return nil, err } @@ -199,7 +220,7 @@ func NewVideoTrackStreamProducer(broadcast *moq.BroadcastProducer, target Target if err != nil { return nil, err } - inner, err := ffi.MoqMediaTrackStreamProducerVideo(bridge.BroadcastProducer(broadcast), resolved, init) + inner, err := ffi.MoqMediaTrackStreamProducerVideo(bridge.BroadcastProducer(broadcast), resolved, init.ffi()) if err != nil { return nil, err } @@ -391,6 +412,14 @@ func (m *TrackProducer) Discontinuity() error { return m.inner.Discontinuity() } +// SetEnabled advertises whether this video rendition is being produced, as its catalog +// enabled flag. Disable it when its sender stops for lack of bandwidth and mark a +// Discontinuity; enable it again just before the keyframe that resumes it. An audio +// track refuses. +func (m *TrackProducer) SetEnabled(enabled bool) error { + return m.inner.SetEnabled(enabled) +} + // Cut draws a group boundary here. // // Audio has no boundary of its own (every packet is independently decodable), so this is diff --git a/go/wrapper/session.go b/go/wrapper/session.go index c099c4a63da..06f910f851b 100644 --- a/go/wrapper/session.go +++ b/go/wrapper/session.go @@ -176,6 +176,44 @@ func (r *Reservation) Update(maxBps uint64) { r.inner.Update(maxBps) } +// RatePolicy maps a reservation's grant onto a sender's bitrate: MaxBps caps it, +// and LowerBps, the next lower rendition's MaxBps, disables this rendition while +// the grant is below (MaxBps + 2*LowerBps) / 3 and a session probes for it. +type RatePolicy = ffi.MoqRatePolicy + +// RateTarget is what a sender following a RateControl should be doing: a +// RateTargetEnabled bitrate, or RateTargetDisabled. +type RateTarget = ffi.MoqRateTarget + +// RateTargetEnabled produces at BitrateBps and advertises the rendition as enabled. +type RateTargetEnabled = ffi.MoqRateTargetEnabled + +// RateTargetDisabled stops sending and advertises the rendition as disabled. +type RateTargetDisabled = ffi.MoqRateTargetDisabled + +// RateControl turns a Reservation's grant into a sender's RateTarget, with the +// policy moq's own encoders follow: drops apply at once, raises climb at most 25% +// per second, and moves under 5% are ignored. For a sender that encodes on its +// own, call Update on every frame and apply what it returns. +type RateControl struct { + inner *ffi.MoqRateControl +} + +// NewRateControl starts at policy.MaxBps until a grant says otherwise. +func NewRateControl(policy RatePolicy) *RateControl { + return &RateControl{inner: ffi.NewMoqRateControl(policy)} +} + +// Target returns what the sender should currently be doing. +func (c *RateControl) Target() RateTarget { + return c.inner.Target() +} + +// Update feeds reservation's grant, returning a new target worth applying, or nil. +func (c *RateControl) Update(reservation *Reservation) *RateTarget { + return c.inner.Update(reservation.inner) +} + // Publish returns the origin used to advertise local broadcasts to the remote. func (s *Session) Publish() *OriginProducer { return &OriginProducer{inner: s.inner.Publish()} diff --git a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt index 0fd36229c93..fe561d01165 100644 --- a/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt +++ b/kt/moq/src/jvmAndAndroidMain/kotlin/dev/moq/Aliases.kt @@ -283,6 +283,15 @@ typealias BandwidthConsumer = uniffi.moq.MoqBandwidthConsumer /** One track's standing claim on an [Allocator]. */ typealias Reservation = uniffi.moq.MoqReservation +/** Turns a [Reservation]'s grant into a sender's [RateTarget]; call `update` on every frame. */ +typealias RateControl = uniffi.moq.MoqRateControl + +/** How a [RateControl] maps a grant onto a bitrate, and the lower rendition that disables it. */ +typealias RatePolicy = uniffi.moq.MoqRatePolicy + +/** What a sender following a [RateControl] should be doing: an enabled bitrate, or disabled. */ +typealias RateTarget = uniffi.moq.MoqRateTarget + /** A snapshot of transport connection statistics. */ typealias ConnectionStats = uniffi.moq.MoqConnectionStats diff --git a/py/moq-rs/moq/__init__.py b/py/moq-rs/moq/__init__.py index 257c0e854f3..1725cd51a03 100644 --- a/py/moq-rs/moq/__init__.py +++ b/py/moq-rs/moq/__init__.py @@ -41,7 +41,15 @@ TrackRequests, ) from .server import Request, Server, Transport -from .session import Allocator, BandwidthConsumer, GrantConsumer, Reservation, Session, Tokens +from .session import ( + Allocator, + BandwidthConsumer, + GrantConsumer, + RateControl, + Reservation, + Session, + Tokens, +) from .subscribe import ( BroadcastConsumer, GroupConsumer, @@ -69,6 +77,7 @@ ProtocolKind, QuicConfig, Range, + RateTarget, Route, Subscription, Timestamp, @@ -131,6 +140,8 @@ "Position", "Publication", "Range", + "RateControl", + "RateTarget", "Request", "Reservation", "Route", diff --git a/py/moq-rs/moq/media.py b/py/moq-rs/moq/media.py index d82b757c691..106277a431b 100644 --- a/py/moq-rs/moq/media.py +++ b/py/moq-rs/moq/media.py @@ -2,6 +2,7 @@ from __future__ import annotations +import copy import json from dataclasses import dataclass, field from typing import Any, cast @@ -167,9 +168,20 @@ def audio(cls, broadcast: BroadcastProducer, init: AudioInit, *, target: Target return cls(MoqMediaTrackProducer.audio(broadcast._inner, target._ffi(), init)) @classmethod - def video(cls, broadcast: BroadcastProducer, init: VideoInit, *, target: Target = Named()) -> TrackProducer: - """Import complete encoded video frames on a named or requested track.""" - return cls(MoqMediaTrackProducer.video(broadcast._inner, target._ffi(), init)) + def video( + cls, + broadcast: BroadcastProducer, + init: VideoInit, + *, + target: Target = Named(), + lower: TrackProducer | None = None, + ) -> TrackProducer: + """Import complete encoded video frames on a named or requested track. + + Pass ``lower``, the next lower rendition of the same source, to publish this + track one priority below it, so a bandwidth allocator funds that one first. + """ + return cls(MoqMediaTrackProducer.video(broadcast._inner, target._ffi(), _lower(init, lower))) def __init__(self, inner: MoqMediaTrackProducer) -> None: self._inner = inner @@ -186,6 +198,15 @@ def discontinuity(self) -> None: """Mark a timeline break, preserving the advertised jitter and delay.""" self._inner.discontinuity() + def set_enabled(self, enabled: bool) -> None: + """Advertise whether this video rendition is being produced, as its catalog ``enabled``. + + Disable it when its sender stops for lack of bandwidth and mark a + :meth:`discontinuity`; enable it again just before the keyframe that + resumes it. An audio track refuses. + """ + self._inner.set_enabled(enabled) + def cut(self) -> None: """Draw a group boundary here. @@ -253,9 +274,19 @@ class TrackStreamProducer: """ @classmethod - def video(cls, broadcast: BroadcastProducer, init: VideoInit, *, target: Target = Named()) -> TrackStreamProducer: - """Import a video byte stream, inferring frame boundaries.""" - return cls(MoqMediaTrackStreamProducer.video(broadcast._inner, target._ffi(), init)) + def video( + cls, + broadcast: BroadcastProducer, + init: VideoInit, + *, + target: Target = Named(), + lower: TrackProducer | None = None, + ) -> TrackStreamProducer: + """Import a video byte stream, inferring frame boundaries. + + ``lower`` is as for :meth:`TrackProducer.video`. + """ + return cls(MoqMediaTrackStreamProducer.video(broadcast._inner, target._ffi(), _lower(init, lower))) def __init__(self, inner: MoqMediaTrackStreamProducer) -> None: self._inner = inner @@ -273,6 +304,14 @@ def finish(self) -> None: self._inner.finish() +def _lower(init: VideoInit, lower: TrackProducer | None) -> VideoInit: + if lower is None: + return init + init = copy.copy(init) + init.lower = lower._inner + return init + + def _rendition(id: str, rendition: Video | Audio) -> MoqRendition: if isinstance(rendition, Video): return cast(MoqRendition, MoqRendition.VIDEO(id=id, config=rendition)) diff --git a/py/moq-rs/moq/session.py b/py/moq-rs/moq/session.py index 900877ca1f6..6c1670223ea 100644 --- a/py/moq-rs/moq/session.py +++ b/py/moq-rs/moq/session.py @@ -8,6 +8,8 @@ MoqAllocator, MoqBandwidthConsumer, MoqGrantConsumer, + MoqRateControl, + MoqRatePolicy, MoqReservation, MoqSession, MoqTokens, @@ -17,7 +19,7 @@ from .errors import is_shutdown from .origin import OriginConsumer, OriginProducer from .publish import TrackDemand -from .types import ConnectionState, ConnectionStats, Grant +from .types import ConnectionState, ConnectionStats, Grant, RateTarget class Session: @@ -268,3 +270,29 @@ def rate(self) -> int | None: def update(self, max_bps: int) -> None: """Change the ceiling, keeping the same claim.""" self._inner.update(max_bps) + + +class RateControl: + """Turns a :class:`Reservation`'s grant into a sender's :data:`RateTarget`. + + The same policy moq's own encoders follow: drops apply at once, raises climb + at most 25% per second, and moves under 5% are ignored. For a sender that + encodes on its own, such as a pre-encoded track: call :meth:`update` on every + frame and apply what it returns. + + Pass ``lower_bps``, the next lower rendition's ``max_bps``, for a rendition + encoded beside lower ones of the same source: a grant below + ``(max_bps + 2 * lower_bps) / 3`` then disables it while a session probes for + the reservation. + """ + + def __init__(self, max_bps: int, *, lower_bps: int | None = None) -> None: + self._inner = MoqRateControl(MoqRatePolicy(max_bps=max_bps, lower_bps=lower_bps)) + + def target(self) -> RateTarget: + """What the sender should currently be doing.""" + return self._inner.target() + + def update(self, reservation: Reservation) -> RateTarget | None: + """Feed ``reservation``'s grant, returning a new target worth applying, or ``None``.""" + return self._inner.update(reservation._inner) diff --git a/py/moq-rs/moq/types.py b/py/moq-rs/moq/types.py index ff6dc5dc3dc..ae87adc4a34 100644 --- a/py/moq-rs/moq/types.py +++ b/py/moq-rs/moq/types.py @@ -40,6 +40,9 @@ from moq_ffi import ( MoqProtocolKind as ProtocolKind, ) +from moq_ffi import ( + MoqRateTarget as RateTarget, +) from moq_ffi import ( MoqRoute as Route, ) @@ -101,6 +104,7 @@ "ProtocolKind", "QuicConfig", "Range", + "RateTarget", "Route", "Subscription", "Timestamp", diff --git a/quest/m1/README.md b/quest/m1/README.md index 2cfb0797089..9d67d073196 100644 --- a/quest/m1/README.md +++ b/quest/m1/README.md @@ -158,7 +158,7 @@ per-frame CPU growth since #5926. - [A moq-transport request token is one auth per distinct token](/quest/m1/ietf-token-auth.md) - requests naming the same token share one verification and grant, on any request type, ending with the last such request - [moq-quic's API improves on quinn's](/quest/m1/quic-api/README.md) - moq-quic and both QUIC drivers ship one mirrored API: `send::`/`recv::` namespaces, one error per direction, consuming terminal operations, plain configs, no quinn baggage - [On-demand JavaScript FETCH](/quest/m1/js-fetch-on-demand.md) - browser publishers answer IETF FETCH for evicted or uncached groups through the JS ranges request surface -- [Native rung disable](/quest/m1/simulcast-rung-disable.md) - OBS stops sending the top renditions its grant cannot fund, as the Rust/FFI publishers do, and a shaped uplink through moq-ffi proves the resume +- [Native rung disable proof](/quest/m1/simulcast-rung-disable.md) - a shaped uplink through moq-ffi proves the top rendition stops and resumes over real BBR3 probing - [Audio jitter target](/quest/m1/audio-jitter-target/README.md) - the playout target is default in both languages; a manual browser proof and a native trace replay remain - [A/V clock](/quest/m1/av-clock.md) - the audio playhead drives Sync.reference while audio plays, through per-track sync handles - [Container reads drop backfill](/quest/m1/read-backfill.md) - the JS container consumer drops groups below its read cursor like Rust diff --git a/quest/m1/simulcast-rung-disable.md b/quest/m1/simulcast-rung-disable.md index 28c72ae0a9f..4d7a25fad3d 100644 --- a/quest/m1/simulcast-rung-disable.md +++ b/quest/m1/simulcast-rung-disable.md @@ -1,40 +1,29 @@ -# [M] OBS disables simulcast rungs the grant cannot fund, proven on a shaped uplink +# [M] A shaped uplink proves native rung disable and resume ## Goal -The OBS plugin stops sending its top video renditions when their bandwidth -grant cannot fund them, advertising each as `enabled: false`, and re-enables -them with hysteresis once the grant recovers, as the Rust and FFI publishers -already do. One shaped-uplink run through moq-ffi proves the rule end to end -over real BBR3 probing, including resuming a rung after the grant fell below -the lowest rung. The JS publisher is +One shaped-uplink run through moq-ffi proves the rung rule end to end over +real BBR3 probing: as the bottleneck narrows below a rung's band, the top +rendition stops and is advertised `enabled: false`, and once the bottleneck +opens back up it resumes with hysteresis, including after the grant fell below +the lowest rung. It runs nightly. The JS publisher is [JS rung disable](/quest/m2/probe/rung-disable-js.md). ## Plan -Landed in [moq-dev/moq#6781](https://github.com/moq-dev/moq/pull/6781): the -rule lives in `moq_mux::rate` (`Policy::lower` sets the band boundary -`(max + 2 * lower) / 3`, `Follower` yields `rate::Target::Disabled`), only -once a session probes for the reservation (`bandwidth::Consumer::probed`). -An idle reservation reads the share it would get, so recovery does not depend -on demand. The FFI video producer takes `lower` (the next rendition down), -sits one track priority below it, drains its encoder while disabled, and -reopens on a keyframe; `moq_encode_video` takes the same `lower` handle in -its `moq_video_encode_options`. +The rule and every native publisher landed: `moq_mux::rate` in +[moq-dev/moq#6781](https://github.com/moq-dev/moq/pull/6781) (`Policy::lower` +sets the band boundary `(max + 2 * lower) / 3`; a rung disables only while a +session probes for its reservation), the FFI and C video producers there too, +and OBS through moq-ffi's `MoqRateControl`, `MoqVideoInit.lower`, and +`MoqMediaTrackProducer::set_enabled`, which a pre-encoded publisher follows +the same way. -- OBS: each OBS video encoder is its own rendition (`video_tracks` in - `cpp/obs/src/moq-output.cpp`) publishing pre-encoded packets through a - media track, and `MoQRate` (`cpp/obs/src/moq-rate.h`) mirrors - `rate::Control` without the rung rule. It needs, through the generated C++ - package: a way to flip a media track's catalog `enabled`, the reservation's - `probed` state (dropped from `MoqReservation` in #6781 for lack of a - consumer), and the lower encoder's ceiling and priority. OBS cannot stop an - encoder or force a keyframe per rung mid-stream, so decide between dropping - packets while disabled and resuming at the next natural keyframe, or - restarting the encoder. -- Shaped uplink: a `moq-shaper` profile whose rate steps down and back up - (like the `cellular-step` built-in) opens the bottleneck back up mid-run. - Run it nightly with an `#[ignore]` test and a recipe in `rs/justfile`, +- Shape it with a `moq-shaper` profile whose rate steps down below the top + rung and back up (like the `cellular-step` built-in), behind a deep but + finite queue whose overflow drops packets. Assert the disable, the catalog + `enabled` flips, and the resume, not exact timings. +- Run it nightly with an `#[ignore]` test and a recipe in `rs/justfile`, beside `quic-sim`. ## Related diff --git a/quest/m2/probe/README.md b/quest/m2/probe/README.md index 8ebcdd6ed79..540d2a9bca5 100644 --- a/quest/m2/probe/README.md +++ b/quest/m2/probe/README.md @@ -83,7 +83,7 @@ This README's own work, once the children land: ## Required -- [Native rung disable](/quest/m1/simulcast-rung-disable.md) - OBS stops sending the top renditions its grant cannot fund, as the Rust/FFI publishers do, and a shaped uplink through moq-ffi proves the resume +- [Native rung disable proof](/quest/m1/simulcast-rung-disable.md) - a shaped uplink through moq-ffi proves the top rendition stops and resumes over real BBR3 probing - [Browser send estimate](/quest/m2/probe/browser-send-estimate.md) - shipping Chrome and Safari publishers adapt to the relay's PROBE estimate, since they have no `getStats()` - [Padding probes](/quest/m2/probe/padding.md) - a browser probes with a lowest-priority padding stream on an ALR schedule - [JS rung disable](/quest/m2/probe/rung-disable-js.md) - `@moq/publish` applies the native rung rule, recovering through padding diff --git a/rs/moq-ffi/src/bandwidth.rs b/rs/moq-ffi/src/bandwidth.rs index beed37ae3a1..3deefa4af9f 100644 --- a/rs/moq-ffi/src/bandwidth.rs +++ b/rs/moq-ffi/src/bandwidth.rs @@ -88,6 +88,84 @@ impl MoqReservation { } } +/// How a [`MoqRateControl`] maps a reservation's grant onto the bitrate a sender produces. +#[derive(Clone, Debug, uniffi::Record)] +pub struct MoqRatePolicy { + /// The most the sender produces, normally its configured bitrate. The grant only lowers it. + pub max_bps: u64, + /// The next lower rendition's `max_bps`, when this one is encoded beside lower renditions + /// of the same source. A grant below `(max_bps + 2 * lower_bps) / 3` then disables this + /// one, since the lower rendition serves viewers better than a starved copy of this one. + /// Absent for the lowest or only rendition, which never disables. + #[uniffi(default = None)] + pub lower_bps: Option, +} + +/// What a sender following a [`MoqRateControl`] should be doing. +#[derive(Clone, Copy, Debug, PartialEq, Eq, uniffi::Enum)] +pub enum MoqRateTarget { + /// Produce at this bitrate, and advertise the rendition as enabled. + Enabled { + /// The bitrate to produce at, in bits per second. + bitrate_bps: u64, + }, + /// Stop sending, and advertise the rendition as disabled until a later target enables it. + Disabled, +} + +impl From for MoqRateTarget { + fn from(target: moq_mux::rate::Target) -> Self { + match target { + moq_mux::rate::Target::Enabled(rate) => Self::Enabled { + bitrate_bps: rate.as_bps(), + }, + moq_mux::rate::Target::Disabled => Self::Disabled, + } + } +} + +/// Turns a reservation's grant into a sender's target, the way moq-ffi's own encoders follow it. +/// +/// Drops apply at once, raises climb at most 25% per second, and moves under 5% are ignored, +/// so a jittering grant doesn't retune the sender on every frame. For a sender that encodes +/// on its own, such as a pre-encoded track whose encoder the caller retunes: call +/// [`update`](Self::update) on every frame and apply what it returns. +#[derive(uniffi::Object)] +pub struct MoqRateControl { + inner: std::sync::Mutex, +} + +#[uniffi::export] +impl MoqRateControl { + /// Start at the policy's `max_bps`, until a grant says otherwise. + #[uniffi::constructor] + pub fn new(policy: MoqRatePolicy) -> Arc { + let mut native = moq_mux::rate::Policy::new(moq_net::bandwidth::Rate::from_bps(policy.max_bps)); + native.lower = policy.lower_bps.map(moq_net::bandwidth::Rate::from_bps); + Arc::new(Self { + inner: std::sync::Mutex::new(moq_mux::rate::Control::new(native)), + }) + } + + /// What the sender should currently be doing. + pub fn target(&self) -> MoqRateTarget { + self.inner.lock().unwrap().target().into() + } + + /// Feed `reservation`'s current grant, returning the new target when it moved enough to + /// apply, or `None` to keep doing what the last target said. + /// + /// A rendition only disables while a session probes for the reservation, since nothing + /// else could raise the grant to resume it. A missing grant (no estimate) holds the target. + pub fn update(&self, reservation: &MoqReservation) -> Option { + let _guard = crate::ffi::enter(); + let grant = reservation.inner.consume(); + let mut control = self.inner.lock().unwrap(); + control.set_probed(grant.probed()); + control.update(grant.peek(), kio::clock::now()).map(Into::into) + } +} + #[cfg(test)] mod tests { use super::*; @@ -203,6 +281,39 @@ mod tests { assert_eq!(reserved.rate(), Some(1_000_000)); } + /// A rung follows its reservation's grant down, but never disables while no session + /// probes for it, since nothing could then raise the grant to resume it. A missing + /// grant holds the target. + #[test] + fn a_rate_control_disables_only_while_probed() { + let estimate = moq_net::bandwidth::Producer::new(); + let bandwidth = allocator(&estimate); + let broadcast = MoqBroadcastProducer::new().unwrap(); + + let track = named_track(&broadcast, "a"); + let _sub = track.consume(None).unwrap(); + let reserved = bandwidth.reserve(&track.demand().unwrap(), 6_000_000); + let control = MoqRateControl::new(MoqRatePolicy { + max_bps: 6_000_000, + lower_bps: Some(2_000_000), + }); + assert_eq!(control.target(), MoqRateTarget::Enabled { bitrate_bps: 6_000_000 }); + assert_eq!(control.update(&reserved), None); + + // Below the 3.33 Mbps boundary, but unprobed: follow it like any sender. + estimate + .set(Some(moq_net::bandwidth::Rate::from_bps(3_000_000))) + .unwrap(); + assert_eq!( + control.update(&reserved), + Some(MoqRateTarget::Enabled { bitrate_bps: 3_000_000 }) + ); + + estimate.set(None).unwrap(); + assert_eq!(control.update(&reserved), None); + assert_eq!(control.target(), MoqRateTarget::Enabled { bitrate_bps: 3_000_000 }); + } + #[cfg(feature = "video")] #[test] fn a_share_wakes_a_parked_thread() { diff --git a/rs/moq-ffi/src/media.rs b/rs/moq-ffi/src/media.rs index ef2d5da484c..923d794dabb 100644 --- a/rs/moq-ffi/src/media.rs +++ b/rs/moq-ffi/src/media.rs @@ -395,6 +395,11 @@ pub struct MoqVideoInit { /// Catalog fields the stream cannot reveal itself. #[uniffi(default = None)] pub hint: Option, + /// The next lower video rendition of the same source, when this one is published beside + /// it. This track then sits one priority below it, so a bandwidth allocator funds the + /// lower rendition first. + #[uniffi(default = None)] + pub lower: Option>, } /// Shares one source's timestamp offset across containers without keeping its broadcast open. @@ -556,14 +561,23 @@ impl From for moq_mux::import::AudioInit { } } -impl From for moq_mux::import::VideoInit { - fn from(init: MoqVideoInit) -> Self { - let mut out = moq_mux::import::VideoInit::new(init.format.into(), init.data); - out.label = init.label; - if let Some(hint) = init.hint { +impl MoqVideoInit { + /// The importer's init, at one priority below the [`lower`](Self::lower) rendition's. + fn resolve(self) -> Result { + let mut out = moq_mux::import::VideoInit::new(self.format.into(), self.data); + out.label = self.label; + if let Some(hint) = self.hint { out.hint = hint.into(); } - out + if let Some(lower) = self.lower { + let guard = lower.inner.lock().unwrap(); + let below = guard.as_ref().ok_or(MoqError::Closed)?; + let priority = below.video.ok_or_else(|| MoqError::Codec { + reason: "the lower rendition is not a video track".into(), + })?; + out.priority = priority.saturating_sub(1); + } + Ok(out) } } @@ -864,6 +878,8 @@ impl MoqMediaCatalogProducer { struct MediaProducer { // Boxed because the codec splitters/imports make this much larger than the container one. import: Box, + /// A video track's priority, which a higher rung takes one below; `None` for audio. + video: Option, } /// An importer for a container, which may publish several tracks. @@ -975,7 +991,7 @@ impl MoqMediaTrackProducer { reason: format!("init failed: {err}"), } })?; - MoqMediaTrackProducer::from_import(import, &state.children) + MoqMediaTrackProducer::from_import(import, None, &state.children) }) } } @@ -993,8 +1009,9 @@ impl MoqMediaTrackProducer { init: MoqVideoInit, ) -> Result, MoqError> { let _guard = crate::ffi::enter(); + let init = init.resolve()?; broadcast.with_state(|state| { - let init: moq_mux::import::VideoInit = init.into(); + let priority = init.priority; let request = target.resolve(&state.broadcast, &init.format)?; let import = moq_mux::import::Track::video(request, state.catalog.reserve(), init).map_err(|err| { @@ -1002,7 +1019,7 @@ impl MoqMediaTrackProducer { reason: format!("init failed: {err}"), } })?; - MoqMediaTrackProducer::from_import(import, &state.children) + MoqMediaTrackProducer::from_import(import, Some(priority), &state.children) }) } } @@ -1045,8 +1062,8 @@ impl MoqMediaTrackStreamProducer { init: MoqVideoInit, ) -> Result, MoqError> { let _guard = crate::ffi::enter(); + let init = init.resolve()?; broadcast.with_state(|state| { - let init: moq_mux::import::VideoInit = init.into(); let request = target.resolve(&state.broadcast, &init.format)?; let import = @@ -1067,11 +1084,16 @@ impl MoqMediaTrackStreamProducer { } impl MoqMediaTrackProducer { - /// Wrap a single-codec importer. - fn from_import(import: moq_mux::import::Track, children: &Children) -> Result, MoqError> { + /// Wrap a single-codec importer, capturing the priority a video track was published at. + fn from_import( + import: moq_mux::import::Track, + video: Option, + children: &Children, + ) -> Result, MoqError> { children.adopt(Arc::new(Self { inner: std::sync::Mutex::new(Some(MediaProducer { import: Box::new(import), + video, })), children: children.clone(), })) @@ -1128,6 +1150,24 @@ impl MoqMediaTrackProducer { Ok(()) } + /// Advertise whether this video rendition is being produced, as its catalog `enabled`. + /// + /// Disable it when its sender stops for lack of bandwidth (a + /// [`MoqRateTarget::Disabled`](crate::bandwidth::MoqRateTarget::Disabled)) and mark a + /// [`discontinuity`](Self::discontinuity), so viewers select a lower rendition instead of + /// waiting for frames. Enable it again just before writing the keyframe that resumes it. + /// The rest of the rendition, its maximum bitrate included, stays as advertised. An audio + /// track refuses. + pub fn set_enabled(&self, enabled: bool) -> Result<(), MoqError> { + let _guard = crate::ffi::enter(); + let mut guard = self.inner.lock().unwrap(); + let media = guard.as_mut().ok_or(MoqError::Closed)?; + media.import.set_enabled(enabled).map_err(|err| MoqError::Codec { + reason: format!("set enabled failed: {err}"), + })?; + Ok(()) + } + /// Draw a group boundary here. /// /// Audio has no boundary of its own (every packet is independently decodable), so this is the diff --git a/rs/moq-ffi/src/test.rs b/rs/moq-ffi/src/test.rs index b3879aae23d..169cc9bdfe0 100644 --- a/rs/moq-ffi/src/test.rs +++ b/rs/moq-ffi/src/test.rs @@ -227,6 +227,7 @@ fn video_init(format: MoqVideoFormat, data: Vec) -> MoqVideoInit { data, label: None, hint: None, + lower: None, } } @@ -2907,6 +2908,51 @@ async fn video_publish_named_track() { assert!(catalog.video.contains_key("hd"), "catalog: {:?}", catalog.video.keys()); } +/// A pre-encoded rung above a lower rendition sits one priority below it, so an allocator +/// funds the lower one first, and disables in the catalog without leaving it. +#[tokio::test] +async fn video_rung_sits_below_its_lower_and_disables() { + let broadcast = MoqBroadcastProducer::new().unwrap(); + let consumer = broadcast.consume().unwrap(); + let catalog_consumer = MoqMediaCatalogConsumer::subscribe(&consumer).await.unwrap(); + + let named = |name: &str| MoqMediaTarget::Named { + name: Some(name.into()), + }; + let low = + MoqMediaTrackProducer::video(&broadcast, named("low"), video_init(MoqVideoFormat::Avc3, h264_init())).unwrap(); + let high = MoqMediaTrackProducer::video( + &broadcast, + named("high"), + MoqVideoInit { + lower: Some(low.clone()), + ..video_init(MoqVideoFormat::Avc3, h264_init()) + }, + ) + .unwrap(); + + let priority = |name: &str| { + let consumer = consumer.clone(); + let name = name.to_string(); + async move { consumer.subscribe_track(name, None).await.unwrap().info().priority } + }; + assert_eq!(priority("low").await, hang::catalog::PRIORITY.video); + assert_eq!(priority("high").await, hang::catalog::PRIORITY.video - 1); + + high.set_enabled(false).unwrap(); + let disabled = loop { + let catalog = kio::clock::timeout(TIMEOUT, catalog_consumer.next()) + .await + .expect("timed out waiting for catalog") + .unwrap() + .expect("expected a catalog"); + if catalog.video.get("high").is_some_and(|high| !high.enabled) { + break catalog; + } + }; + assert!(disabled.video["low"].enabled); +} + /// A keyframe whose SPS names another level is a new decoder identity, so the track producer moves /// to a new track, and its demand follows rather than watching the one that ended. #[tokio::test] diff --git a/rs/moq-mux/src/codec/av1/import.rs b/rs/moq-mux/src/codec/av1/import.rs index e045c515df0..6d2d960b7b2 100644 --- a/rs/moq-mux/src/codec/av1/import.rs +++ b/rs/moq-mux/src/codec/av1/import.rs @@ -216,6 +216,13 @@ impl Import { self.track.discontinuity() } + /// Advertise whether this rendition is being produced, as the catalog's + /// [`enabled`](hang::catalog::VideoConfig::enabled): a publisher that stops encoding it + /// disables it, so viewers select another rendition rather than wait for frames. + pub fn set_enabled(&mut self, enabled: bool) -> Result<()> { + self.catalog.set_enabled(&mut self.track, enabled) + } + /// Close the current group and open the next one at `sequence`. pub fn seek(&mut self, sequence: u64) -> Result<()> { self.track.seek(sequence)?; diff --git a/rs/moq-mux/src/codec/vp8/import.rs b/rs/moq-mux/src/codec/vp8/import.rs index 7a20db2a6b4..df6afb36b8a 100644 --- a/rs/moq-mux/src/codec/vp8/import.rs +++ b/rs/moq-mux/src/codec/vp8/import.rs @@ -118,6 +118,13 @@ impl Import { self.track.discontinuity() } + /// Advertise whether this rendition is being produced, as the catalog's + /// [`enabled`](hang::catalog::VideoConfig::enabled): a publisher that stops encoding it + /// disables it, so viewers select another rendition rather than wait for frames. + pub fn set_enabled(&mut self, enabled: bool) -> crate::Result<()> { + self.catalog.set_enabled(&mut self.track, enabled) + } + /// Close the current group and open the next one at `sequence`. pub fn seek(&mut self, sequence: u64) -> crate::Result<()> { self.track.seek(sequence)?; diff --git a/rs/moq-mux/src/codec/vp9/import.rs b/rs/moq-mux/src/codec/vp9/import.rs index e851a6a9665..33cf292665c 100644 --- a/rs/moq-mux/src/codec/vp9/import.rs +++ b/rs/moq-mux/src/codec/vp9/import.rs @@ -119,6 +119,13 @@ impl Import { self.track.discontinuity() } + /// Advertise whether this rendition is being produced, as the catalog's + /// [`enabled`](hang::catalog::VideoConfig::enabled): a publisher that stops encoding it + /// disables it, so viewers select another rendition rather than wait for frames. + pub fn set_enabled(&mut self, enabled: bool) -> crate::Result<()> { + self.catalog.set_enabled(&mut self.track, enabled) + } + /// Close the current group and open the next one at `sequence`. pub fn seek(&mut self, sequence: u64) -> crate::Result<()> { self.track.seek(sequence)?; diff --git a/rs/moq-mux/src/import/init.rs b/rs/moq-mux/src/import/init.rs index 9caec7d87e4..dcec2396b89 100644 --- a/rs/moq-mux/src/import/init.rs +++ b/rs/moq-mux/src/import/init.rs @@ -56,6 +56,11 @@ pub struct VideoInit { pub label: Option, /// Catalog fields the stream cannot reveal itself. pub hint: VideoHint, + /// The track's priority, which a [`bandwidth::Allocator`](moq_net::bandwidth::Allocator) + /// funds in order. A rendition published above a lower one of the same source sits one + /// below it, so the lower one is funded first. Defaults to + /// [`PRIORITY.video`](hang::catalog::PRIORITY). + pub priority: i8, } impl VideoInit { @@ -66,6 +71,7 @@ impl VideoInit { data: data.into(), label: None, hint: VideoHint::default(), + priority: hang::catalog::PRIORITY.video, } } } diff --git a/rs/moq-mux/src/import/track.rs b/rs/moq-mux/src/import/track.rs index 0e8f58d7609..dd63d255a6d 100644 --- a/rs/moq-mux/src/import/track.rs +++ b/rs/moq-mux/src/import/track.rs @@ -138,6 +138,15 @@ fn build_av1( Ok((split, import)) } +/// An audio format handed to a video-only operation. +fn not_video(format: AudioFormat) -> crate::Error { + crate::Error::WrongKind { + format: format.to_string(), + actual: "audio", + wanted: "video", + } +} + enum TrackKind { /// H.264 avc3 (Annex-B, inline SPS/PPS). The split owns byte parsing; the /// import publishes. @@ -272,11 +281,7 @@ impl Track { ) -> Result { use hang::catalog::VideoCodec; - let track = request.accept(track_info( - &reserved, - hang::catalog::PRIORITY.video, - &init.hint.container, - )?); + let track = request.accept(track_info(&reserved, init.priority, &init.hint.container)?); let data = init.data.as_ref(); let kind = match init.format { VideoFormat::Avc1 => { @@ -461,6 +466,30 @@ impl Track { import!(&mut self.kind, import => import.discontinuity()) } + /// Advertise whether this video rendition is being produced, as the catalog's + /// [`enabled`](hang::catalog::VideoConfig::enabled). + /// + /// Disable it when the sender stops for lack of bandwidth (a + /// [`rate::Target::Disabled`](crate::rate::Target::Disabled)) and mark the + /// [`discontinuity`](Self::discontinuity), so viewers select a lower rendition instead + /// of waiting for frames. Enable it again with the keyframe that resumes it. Audio has + /// no lower rendition to fall back to, so an audio track refuses with + /// [`Error::WrongKind`](crate::Error::WrongKind). + pub fn set_enabled(&mut self, enabled: bool) -> Result<()> { + match &mut self.kind { + TrackKind::Avc3 { import, .. } | TrackKind::Avc1 { import, .. } => import.set_enabled(enabled)?, + TrackKind::Hev1 { import, .. } | TrackKind::Hvc1 { import, .. } => import.set_enabled(enabled)?, + TrackKind::Av01 { import, .. } => import.set_enabled(enabled)?, + TrackKind::Vp8(import) => import.set_enabled(enabled)?, + TrackKind::Vp9(import) => import.set_enabled(enabled)?, + TrackKind::Aac(_) => return Err(not_video(AudioFormat::Aac)), + TrackKind::Opus(_) => return Err(not_video(AudioFormat::Opus)), + TrackKind::Mp3(_) => return Err(not_video(AudioFormat::Mp3)), + TrackKind::Flac(_) => return Err(not_video(AudioFormat::Flac)), + } + Ok(()) + } + /// Close the current group and open the next one at `sequence`. pub fn seek(&mut self, sequence: u64) -> Result<()> { self.group_start = None; @@ -576,11 +605,7 @@ impl TrackStream { reserved: crate::catalog::Reserved, init: VideoInit, ) -> Result { - let track = request.accept(track_info( - &reserved, - hang::catalog::PRIORITY.video, - &init.hint.container, - )?); + let track = request.accept(track_info(&reserved, init.priority, &init.hint.container)?); let hint = video_hint(&init, None); // Only the self-delimiting codecs can be recovered from a raw byte stream. let kind = match init.format { @@ -1156,6 +1181,49 @@ mod tests { let video = consumer.track("video").unwrap().info().await.unwrap(); assert_eq!(video.priority, PRIORITY.video); + + // A higher rung asks for its own priority, below the rendition it sits above. + let request = broadcast.reserve_track("top").unwrap(); + let _top = Track::video( + request, + catalog.reserve(), + VideoInit { + priority: PRIORITY.video - 1, + ..VideoInit::new(VideoFormat::Vp8, Vec::new()) + }, + ) + .unwrap(); + let top = consumer.track("top").unwrap().info().await.unwrap(); + assert_eq!(top.priority, PRIORITY.video - 1); + } + + /// Disabling a video rendition flips its catalog `enabled` and keeps the rest of it; + /// audio has no lower rendition to fall back to, so it refuses. + #[tokio::test(start_paused = true)] + async fn video_disables_and_audio_refuses() { + let (broadcast, catalog) = new_broadcast(); + let request = broadcast.reserve_track("video").unwrap(); + let mut video = Track::video(request, catalog.reserve(), VideoInit::new(VideoFormat::Vp8, Vec::new())).unwrap(); + + video.set_enabled(false).unwrap(); + let rendition = catalog.snapshot().video.renditions.get("video").cloned().unwrap(); + assert!(!rendition.enabled); + assert_eq!(rendition.codec.to_string(), "vp8"); + + video.set_enabled(true).unwrap(); + assert!(catalog.snapshot().video.renditions["video"].enabled); + + let request = broadcast.reserve_track("audio").unwrap(); + let mut audio = Track::audio( + request, + catalog.reserve(), + AudioInit::new(AudioFormat::Opus, opus_head()), + ) + .unwrap(); + assert!(matches!( + audio.set_enabled(false), + Err(crate::Error::WrongKind { actual: "audio", .. }) + )); } /// An audio format with no init bytes errors up front (audio can't resolve its config from frames), diff --git a/sh/obs/unit.sh b/sh/obs/unit.sh index 6a21fb5e4a9..46ba0649fe7 100755 --- a/sh/obs/unit.sh +++ b/sh/obs/unit.sh @@ -64,7 +64,7 @@ for name in moq-audio-encoder moq-output moq-settings moq-source obs-moq; do -o "$out/$name-test" "${sources[@]}" "$out/moq.o" $moq_libs TSAN_OPTIONS="halt_on_error=1 suppressions=$PWD/test/tsan.supp" "$out/$name-test" done -for name in moq-quality-defaults moq-dock-stop moq-error moq-encoder-latency moq-dial moq-spark moq-rate; do +for name in moq-quality-defaults moq-dock-stop moq-error moq-encoder-latency moq-dial moq-spark; do "$cxx" "${flags[@]}" "${includes[@]}" \ -o "$out/$name-test" "test/$name-test.cpp" TSAN_OPTIONS="halt_on_error=1 suppressions=$PWD/test/tsan.supp" "$out/$name-test" diff --git a/swift/Sources/Moq/Bandwidth.swift b/swift/Sources/Moq/Bandwidth.swift index 917b4199046..71abf536bee 100644 --- a/swift/Sources/Moq/Bandwidth.swift +++ b/swift/Sources/Moq/Bandwidth.swift @@ -58,3 +58,33 @@ public final class Reservation: Sendable { ffi.update(maxBps: maxBps) } } + +/// What a sender following a ``RateControl`` should be doing. +public typealias RateTarget = MoqRateTarget + +/// Turns a ``Reservation``'s grant into a sender's ``RateTarget``. +/// +/// The same policy moq's own encoders follow: drops apply at once, raises climb at most 25% per +/// second, and moves under 5% are ignored. For a sender that encodes on its own, such as a +/// pre-encoded track: call `update(_:)` on every frame and apply what it returns. +/// +/// Pass `lowerBps`, the next lower rendition's `maxBps`, for a rendition encoded beside lower +/// ones of the same source: a grant below `(maxBps + 2 * lowerBps) / 3` then disables it while +/// a session probes for the reservation. +public final class RateControl: Sendable { + let ffi: MoqRateControl + + public init(maxBps: UInt64, lowerBps: UInt64? = nil) { + ffi = MoqRateControl(policy: MoqRatePolicy(maxBps: maxBps, lowerBps: lowerBps)) + } + + /// What the sender should currently be doing. + public func target() -> RateTarget { + ffi.target() + } + + /// Feed `reservation`'s grant, returning a new target worth applying, or nil. + public func update(_ reservation: Reservation) -> RateTarget? { + ffi.update(reservation: reservation.ffi) + } +} diff --git a/swift/Sources/Moq/Media.swift b/swift/Sources/Moq/Media.swift index 87d84360c1e..80fdf822acf 100644 --- a/swift/Sources/Moq/Media.swift +++ b/swift/Sources/Moq/Media.swift @@ -237,10 +237,17 @@ public enum Media { TrackProducer(try MoqMediaTrackProducer.audio(broadcast: broadcast.ffi, target: target.ffi, init: initData)) } /// Import complete video frames on a named or requested track. - public static func video(broadcast: BroadcastProducer, initData: VideoInit, target: Target = .named(name: nil)) - throws -> TrackProducer - { - TrackProducer(try MoqMediaTrackProducer.video(broadcast: broadcast.ffi, target: target.ffi, init: initData)) + /// + /// Pass `lower`, the next lower rendition of the same source, to publish this track one + /// priority below it, so a bandwidth allocator funds that one first. + public static func video( + broadcast: BroadcastProducer, initData: VideoInit, target: Target = .named(name: nil), + lower: TrackProducer? = nil + ) throws -> TrackProducer { + var initData = initData + if let lower { initData.lower = lower.ffi } + return TrackProducer( + try MoqMediaTrackProducer.video(broadcast: broadcast.ffi, target: target.ffi, init: initData)) } init(_ ffi: MoqMediaTrackProducer) { self.ffi = ffi @@ -264,6 +271,14 @@ public enum Media { try ffi.discontinuity() } + /// Advertise whether this video rendition is being produced, as its catalog `enabled`. + /// + /// Disable it when its sender stops for lack of bandwidth and mark a ``discontinuity()``; + /// enable it again just before the keyframe that resumes it. An audio track refuses. + public func setEnabled(_ enabled: Bool) throws { + try ffi.setEnabled(enabled: enabled) + } + /// Draw a group boundary here. /// /// Audio has no boundary of its own (every packet is independently decodable), so this is the @@ -332,10 +347,15 @@ public enum Media { let ffi: MoqMediaTrackStreamProducer /// Import a video byte stream, inferring frame boundaries. - public static func video(broadcast: BroadcastProducer, initData: VideoInit, target: Target = .named(name: nil)) - throws -> TrackStreamProducer - { - TrackStreamProducer( + /// + /// `lower` is as for ``TrackProducer/video(broadcast:initData:target:lower:)``. + public static func video( + broadcast: BroadcastProducer, initData: VideoInit, target: Target = .named(name: nil), + lower: TrackProducer? = nil + ) throws -> TrackStreamProducer { + var initData = initData + if let lower { initData.lower = lower.ffi } + return TrackStreamProducer( try MoqMediaTrackStreamProducer.video(broadcast: broadcast.ffi, target: target.ffi, init: initData)) } init(_ ffi: MoqMediaTrackStreamProducer) {