From c53f0ce556f8c550b72e0144178ca0be6daab6ca Mon Sep 17 00:00:00 2001 From: iflyhere <57563846+iflyhere@users.noreply.github.com> Date: Wed, 2 Sep 2026 20:01:50 +0200 Subject: [PATCH 1/2] Bound how long the packet queue holds a frame back BufferedPacketQueue reorders the RTP stream before it reaches the parser. When a packet is missing it holds everything behind the gap, which is right for a reorder and wrong for a loss - and past wfb-ng, on a loopback socket, a gap is almost always a loss that FEC could not recover. Three things made that wait longer than it needs to be. The monotonic-increase escape hatch, which exists precisely to notice "the sequence numbers keep climbing but the gap is not filling", never fires. calculateDistance(a, b) is how far b is ahead of a - seqLessThan reads it that way - but the call site asks for the distance from the incoming packet to the last delivered one. For a gap ahead of us that is negative, so the else branch clears the counter on every single packet. Fixing the argument order alone is not enough: the counter is only incremented while abs(dist) < MONOTONIC_THRESHOLD, and dist grows by one with every packet held back. Compared against the same constant that gates it, the counter tops out at three and can never reach five. The gate is now its own constant, so a run past a gap releases the buffer after five packets instead of never. That leaves MAX_BUFFER_SIZE as the only bound, and how much latency fifteen packets is depends entirely on the packet rate: about 20ms on a 1080p video stream, but roughly 300ms on the audio stream, which runs at a fraction of it. The buffer is now also bounded in time, so the worst case is the same on both. Twenty milliseconds is a little over one frame at 60fps and several orders of magnitude more than a loopback socket needs to reorder anything. Nothing is dropped by any of this - the buffered packets are still delivered, just without waiting on one that is not coming. The test target could not be built on a case-sensitive filesystem, since CMakeLists.txt spelled the source BufferedPacketqueue_test.cpp. The existing two cases both turn out to feed a strictly in-order stream, so neither reached the buffer at all; the new ones cover a permanent gap, the age bound, a reorder that resolves inside it, and a jump too large to be one. They drive the clock themselves rather than reading it, so the age bound cannot make them flaky on a loaded machine. --- .../src/main/cpp/BufferedPacketQueue.h | 94 +++++++++++++++++-- .../cpp/tests/BufferedPacketQueue_test.cpp | 75 ++++++++++++++- .../src/main/cpp/tests/CMakeLists.txt | 2 +- 3 files changed, 159 insertions(+), 12 deletions(-) diff --git a/app/videonative/src/main/cpp/BufferedPacketQueue.h b/app/videonative/src/main/cpp/BufferedPacketQueue.h index 88d6eab0..5efcc9d1 100644 --- a/app/videonative/src/main/cpp/BufferedPacketQueue.h +++ b/app/videonative/src/main/cpp/BufferedPacketQueue.h @@ -5,9 +5,11 @@ #include #endif #include +#include #include #include #include +#include #include #include @@ -15,10 +17,27 @@ // Define logging tag and maximum buffer size #define BUFFERED_QUEUE_LOG_TAG "BufferedPacketQueue" -// Considering the packet rate about 100 packets per second, 10 packets should be enough +// Last-resort cap on the buffer. How long this is in wall clock time depends entirely on the +// packet rate, which is why it cannot be the only bound - see MAX_BUFFER_AGE. constexpr size_t MAX_BUFFER_SIZE = 15; // Number of monotonically increasing packets constexpr size_t MONOTONIC_THRESHOLD = 5; +// A monotonic run is only tracked while the gap is plausibly a reorder. Kept separate from +// MONOTONIC_THRESHOLD: the gap grows by one with every packet held back, so gating the counter +// on the same value it is compared against means it can never reach it. +constexpr size_t MONOTONIC_MAX_DISTANCE = 64; + +// Clock used to bound how long a packet may be held back. +using QueueClock = std::chrono::steady_clock; +using QueueTimePoint = QueueClock::time_point; + +// A gap in the sequence numbers on this path is almost always a packet that FEC could not +// recover, not a reorder: by the time packets get here they have come through wfb-ng and a +// loopback socket, where reordering takes microseconds. Waiting for a packet that will never +// arrive is pure added latency, so the wait is bounded in time rather than in packets - the +// packet bound alone is worth ~20ms on a 1080p video stream but ~300ms on the audio stream, +// which runs at a fraction of the packet rate. +constexpr auto MAX_BUFFER_AGE = std::chrono::milliseconds(20); // Type definition for sequence numbers using SeqType = uint16_t; @@ -45,7 +64,12 @@ class BufferedPacketQueue * @param callback Callable to handle processed packets. */ template - void processPacket(SeqType currPacketIdx, const uint8_t* data, std::size_t data_length, Callback& callback) + void processPacket( + SeqType currPacketIdx, + const uint8_t* data, + std::size_t data_length, + Callback& callback, + QueueTimePoint now = QueueClock::now()) { logDebug( "Processing packet with Sequence=%u, lastPacketIdx=%u, firstPacket=%s", @@ -53,6 +77,19 @@ class BufferedPacketQueue mLastPacketIdx, mFirstPacket ? "true" : "false"); + // Before anything else: give up on a gap we have been waiting on for too long. Done + // here rather than in handleOutOfOrderPacket so that an in-order packet arriving after + // a stall does not get delivered ahead of what is already buffered. + if (!mPackets.empty() && (now - mOldestBufferedAt) >= MAX_BUFFER_AGE) + { + logWarning( + "Held %zu packet(s) for more than %lldms waiting on Sequence=%u. Flushing.", + mPackets.size(), + (long long) MAX_BUFFER_AGE.count(), + static_cast(static_cast(mLastPacketIdx + 1))); + mLastPacketIdx = drainBufferInOrder(callback); + } + if (isFirstPacket(currPacketIdx)) { handleFirstPacket(currPacketIdx); @@ -73,7 +110,7 @@ class BufferedPacketQueue else { // Out-of-order packet - handleOutOfOrderPacket(currPacketIdx, data, data_length, callback); + handleOutOfOrderPacket(currPacketIdx, data, data_length, callback, now); } } @@ -83,6 +120,9 @@ class BufferedPacketQueue std::unordered_map> mPackets; + // When the current stall started, i.e. when mPackets last went from empty to non-empty. + QueueTimePoint mOldestBufferedAt{}; + // This variable is used to track a situation where the sequence number is increasing monotonically while packets // are out of order. if this counter reaches MONOTONIC_THRESHOLD, we will restart buffering and update lastPacketIdx // to the highest sequence index received. @@ -174,7 +214,12 @@ class BufferedPacketQueue * @param callback Callable to handle processed packets. */ template - void handleOutOfOrderPacket(SeqType currPacketIdx, const uint8_t* data, std::size_t data_length, Callback& callback) + void handleOutOfOrderPacket( + SeqType currPacketIdx, + const uint8_t* data, + std::size_t data_length, + Callback& callback, + QueueTimePoint now) { logDebug("Out-of-order packet detected. Sequence=%u", currPacketIdx); @@ -184,10 +229,15 @@ class BufferedPacketQueue // return; } - bufferPacket(currPacketIdx, data, data_length); + bufferPacket(currPacketIdx, data, data_length, now); - auto dist = calculateDistance(currPacketIdx, mLastPacketIdx); - if (std::abs(dist) < MONOTONIC_THRESHOLD) + // calculateDistance(a, b) is how far b is ahead of a - see seqLessThan below - so the + // question "is this packet ahead of the last one we delivered" has to be asked in that + // order. Reversed, dist is negative for exactly the case this heuristic exists for (a + // gap ahead of us), the else branch below clears the counter every time, and the + // buffer only ever drains on the MAX_BUFFER_SIZE cap. + auto dist = calculateDistance(mLastPacketIdx, currPacketIdx); + if (static_cast(std::abs(dist)) < MONOTONIC_MAX_DISTANCE) { // Check for monotonic increases if (dist > 0) @@ -231,8 +281,15 @@ class BufferedPacketQueue * @param data Pointer to the packet data. * @param data_length Size of the packet data. */ - void bufferPacket(SeqType currPacketIdx, const uint8_t* data, std::size_t data_length) + void bufferPacket(SeqType currPacketIdx, const uint8_t* data, std::size_t data_length, QueueTimePoint now) { + // Only the start of a stall is recorded, because the buffer is always drained as a + // whole. A partial drain leaves the mark where it was, which errs towards flushing + // early - the safe direction on a live link. + if (mPackets.empty()) + { + mOldestBufferedAt = now; + } mPackets[currPacketIdx] = std::vector(data, data + data_length); logDebug("Buffered out-of-order packet. Buffer size: %zu", mPackets.size()); } @@ -244,6 +301,19 @@ class BufferedPacketQueue */ template void restartBuffering(Callback& callback, SeqType currPacketIdx) + { + drainBufferInOrder(callback); + mLastPacketIdx = currPacketIdx; + } + + /** + * @brief Delivers everything currently held back, in sequence order, and empties the buffer. + * @tparam Callback A callable type that processes the packet data. + * @param callback Callable to handle processed packets. + * @return The highest sequence number delivered, or mLastPacketIdx if nothing was held. + */ + template + SeqType drainBufferInOrder(Callback& callback) { // Process as many in-order buffered packets as possible processBufferedPackets(callback); @@ -269,18 +339,24 @@ class BufferedPacketQueue [](const auto& a, const auto& b) { return a->first < b->first; }); // Iterate over the sorted packets and invoke the callback + SeqType highest = mLastPacketIdx; for (const auto& it : sortedPackets) { const auto& packet = it->second; logDebug("Processing possibly out-of-order buffered packet with Sequence=%u.", it->first); callback(packet.data(), packet.size()); + if (calculateDistance(highest, it->first) > 0) + { + highest = it->first; + } } mPackets.clear(); // Reset the monotonic increase counter mMonotonicOutOfOrderIncreaseCount = 0; + return highest; } - mLastPacketIdx = currPacketIdx; + return mLastPacketIdx; } /** diff --git a/app/videonative/src/main/cpp/tests/BufferedPacketQueue_test.cpp b/app/videonative/src/main/cpp/tests/BufferedPacketQueue_test.cpp index c6f79a0b..03ad2d33 100644 --- a/app/videonative/src/main/cpp/tests/BufferedPacketQueue_test.cpp +++ b/app/videonative/src/main/cpp/tests/BufferedPacketQueue_test.cpp @@ -1,6 +1,7 @@ #include "BufferedPacketQueue.h" // the class under test #include #include +#include #include // ---------- Test fixture ---------------------------------------------------- @@ -10,7 +11,17 @@ class BufferedPacketQueueTest : public ::testing::Test BufferedPacketQueue q; std::vector delivered; - void SetUp() override { delivered.clear(); } + // The queue bounds how long it will hold a packet back, so the tests drive the clock + // themselves instead of letting a loaded machine decide whether the bound was hit. + QueueTimePoint now{}; + + void SetUp() override + { + delivered.clear(); + now = QueueTimePoint{}; + } + + void advance(int ms) { now += std::chrono::milliseconds(ms); } /* Helper: feed one packet and record what the queue actually delivers. */ void feed(uint16_t seq) @@ -23,7 +34,7 @@ class BufferedPacketQueueTest : public ::testing::Test delivered.push_back(*(uint16_t*) seq); }; - q.processPacket(seq, (uint8_t*) &dummy, 2, cb); + q.processPacket(seq, (uint8_t*) &dummy, 2, cb, now); } }; @@ -57,6 +68,66 @@ TEST_F(BufferedPacketQueueTest, ReorderedDeliversInOrder) ASSERT_EQ(delivered, expected) << "Overflow flush should deliver the entire block in one shot"; } +// ---------- A gap that will never be filled -------------------------------- +// The common case on a lossy link: FEC could not recover one packet, and every packet after it +// is held back waiting for it. Nothing is lost by waiting, but everything behind the gap gets +// later and later, so the queue has to give up at some point. +TEST_F(BufferedPacketQueueTest, PermanentGapDoesNotHoldTheStreamForFifteenPackets) +{ + for (uint16_t s = 1; s <= 5; ++s) feed(s); + ASSERT_EQ(delivered, (std::vector{1, 2, 3, 4, 5})); + + // 6 never arrives. + for (uint16_t s = 7; s <= 11; ++s) feed(s); + feed(12); + + ASSERT_EQ(delivered, (std::vector{1, 2, 3, 4, 5, 7, 8, 9, 10, 11, 12})) + << "A monotonic run past the gap should release the buffer, not wait for MAX_BUFFER_SIZE"; +} + +// The monotonic run is counted in packets, so how much latency it costs depends on the packet +// rate - which on the audio stream is a fraction of the video one. The age bound is what makes +// the wait the same on both. +TEST_F(BufferedPacketQueueTest, StaleBufferIsFlushedOnTimeout) +{ + feed(1); + // 2 never arrives. + feed(3); + ASSERT_EQ(delivered, (std::vector{1})) << "3 is held back waiting for 2"; + + advance(25); + feed(4); + + ASSERT_EQ(delivered, (std::vector{1, 3, 4})) + << "Once the buffer is older than MAX_BUFFER_AGE it has to be released"; +} + +// The other side of that bound: a reorder that resolves quickly must still be reordered, not +// flushed out of sequence. +TEST_F(BufferedPacketQueueTest, ReorderWithinTimeoutIsStillPutBackInOrder) +{ + feed(1); + feed(3); + advance(5); + feed(2); + + ASSERT_EQ(delivered, (std::vector{1, 2, 3})) + << "A packet that arrives late but within MAX_BUFFER_AGE must not be flushed early"; +} + +// A large jump is not a reorder - it is a stream that restarted somewhere else. It must not be +// mistaken for a monotonic run, and the buffer cap has to catch it. +TEST_F(BufferedPacketQueueTest, LargeJumpFallsBackToTheBufferCap) +{ + feed(1); + for (uint16_t s = 30000; s < 30000 + MAX_BUFFER_SIZE; ++s) feed(s); + + ASSERT_EQ(delivered.size(), 1u + MAX_BUFFER_SIZE) + << "The buffer cap should have released the jumped-to block"; + ASSERT_EQ(delivered.front(), 1); + ASSERT_EQ(delivered.back(), static_cast(30000 + MAX_BUFFER_SIZE - 1)); +} + // ---------- gtest boilerplate main ----------------------------------------- int main(int argc, char** argv) { diff --git a/app/videonative/src/main/cpp/tests/CMakeLists.txt b/app/videonative/src/main/cpp/tests/CMakeLists.txt index 1e70da39..6ce59ad5 100644 --- a/app/videonative/src/main/cpp/tests/CMakeLists.txt +++ b/app/videonative/src/main/cpp/tests/CMakeLists.txt @@ -25,7 +25,7 @@ enable_testing() # ---------- Test executable -------------------------------------------------- add_executable(queue_test - BufferedPacketqueue_test.cpp + BufferedPacketQueue_test.cpp ) target_include_directories(queue_test PUBLIC From ca01d433db95bc98019996814ea0116130c7a626 Mon Sep 17 00:00:00 2001 From: iflyhere <57563846+iflyhere@users.noreply.github.com> Date: Wed, 2 Sep 2026 22:40:01 +0200 Subject: [PATCH 2/2] Advance the queue past everything a flush delivered Two problems with the flush, both found in review. restartBuffering() set mLastPacketIdx to the packet that triggered the flush rather than the newest one delivered. With 1..5 delivered, 6 lost, then 7 8 9 12 10, the flush hands over 7 8 9 10 12 but leaves the pointer on 10 - so 11 would be delivered after 12, and 13..16 are held until 17 arrives. The queue re-created the stall it exists to prevent. bufferPacket() always runs before the flush, so the triggering packet is in the buffer either way and mLastPacketIdx = drainBufferInOrder() is all that was needed; restartBuffering() is gone. That alone would have broken the resync on a large jump, because drainBufferInOrder() seeded `highest` from mLastPacketIdx and compared with calculateDistance(), which reads as negative past half the sequence space. RTP starts at a random sequence number, so a VTX that reboots mid-session lands exactly there: fed 1, 2, then 40000, 40001, ..., neither the seed nor the comparison ever moves and every packet comes out one flush late, forever. On video the fifteen-packet cap used to paper over it by rewinding to currPacketIdx; on audio nothing did. `highest` is now seeded from a packet that is actually in the buffer. While sorting the flush: by raw value a block straddling the wrap point (65534, 65535, 0, 1) sorts to 0, 1, 65534, 65535 and was handed to the parser in that order. Sorting by distance from the last delivered packet fixes the order and makes the last element the newest. Three tests added, one per problem. The large-jump case uses 40000 - the existing LargeJumpFallsBackToTheBufferCap uses 30000, which is still positive as an int16 and does not reach any of this. --- .../src/main/cpp/BufferedPacketQueue.h | 39 ++++++------- .../cpp/tests/BufferedPacketQueue_test.cpp | 58 +++++++++++++++++++ 2 files changed, 75 insertions(+), 22 deletions(-) diff --git a/app/videonative/src/main/cpp/BufferedPacketQueue.h b/app/videonative/src/main/cpp/BufferedPacketQueue.h index 5efcc9d1..891540a8 100644 --- a/app/videonative/src/main/cpp/BufferedPacketQueue.h +++ b/app/videonative/src/main/cpp/BufferedPacketQueue.h @@ -246,10 +246,8 @@ class BufferedPacketQueue logDebug("Monotonic increase count: %zu", mMonotonicOutOfOrderIncreaseCount); if (mMonotonicOutOfOrderIncreaseCount >= MONOTONIC_THRESHOLD) { - restartBuffering(callback, currPacketIdx); - // Update lastPacketIdx to the highest sequence index received - SeqType newLastIdx = currPacketIdx; - logWarning("Monotonic threshold reached. Updating lastPacketIdx to %u", newLastIdx); + mLastPacketIdx = drainBufferInOrder(callback); + logWarning("Monotonic threshold reached. Updating lastPacketIdx to %u", mLastPacketIdx); } } else @@ -264,7 +262,7 @@ class BufferedPacketQueue { logWarning( "Buffer size exceeded MAX_BUFFER_SIZE (%zu). Processing in-order buffered packets.", MAX_BUFFER_SIZE); - restartBuffering(callback, currPacketIdx); + mLastPacketIdx = drainBufferInOrder(callback); } } @@ -294,18 +292,6 @@ class BufferedPacketQueue logDebug("Buffered out-of-order packet. Buffer size: %zu", mPackets.size()); } - /** - * @brief Handles buffer overflow by processing in-order packets and discarding others. - * @tparam Callback A callable type that processes the packet data. - * @param callback Callable to handle processed packets. - */ - template - void restartBuffering(Callback& callback, SeqType currPacketIdx) - { - drainBufferInOrder(callback); - mLastPacketIdx = currPacketIdx; - } - /** * @brief Delivers everything currently held back, in sequence order, and empties the buffer. * @tparam Callback A callable type that processes the packet data. @@ -332,14 +318,23 @@ class BufferedPacketQueue sortedPackets.push_back(it); } - // Sort the vector based on the keys + // Sorted by distance from the last delivered packet, not by raw value: a block + // that straddles the wrap point (65534, 65535, 0, 1) sorts to 0, 1, 65534, 65535 + // by value, and would be handed to the parser in that order. + const SeqType from = mLastPacketIdx; std::sort( sortedPackets.begin(), sortedPackets.end(), - [](const auto& a, const auto& b) { return a->first < b->first; }); - - // Iterate over the sorted packets and invoke the callback - SeqType highest = mLastPacketIdx; + [from](const auto& a, const auto& b) { + return static_cast(a->first - from) < static_cast(b->first - from); + }); + + // Seeded from a packet that is actually in the buffer rather than from + // mLastPacketIdx. RTP starts at a random sequence number, so a VTX that reboots + // mid-session can land more than half the sequence space away, where + // calculateDistance() reads as negative - seeded from mLastPacketIdx nothing + // would ever move and the queue would never resync. + SeqType highest = sortedPackets.front()->first; for (const auto& it : sortedPackets) { const auto& packet = it->second; diff --git a/app/videonative/src/main/cpp/tests/BufferedPacketQueue_test.cpp b/app/videonative/src/main/cpp/tests/BufferedPacketQueue_test.cpp index 03ad2d33..6e803b62 100644 --- a/app/videonative/src/main/cpp/tests/BufferedPacketQueue_test.cpp +++ b/app/videonative/src/main/cpp/tests/BufferedPacketQueue_test.cpp @@ -128,6 +128,64 @@ TEST_F(BufferedPacketQueueTest, LargeJumpFallsBackToTheBufferCap) ASSERT_EQ(delivered.back(), static_cast(30000 + MAX_BUFFER_SIZE - 1)); } +// A flush delivers everything it is holding, so the pointer has to end up past all of it. +// Rewinding it to whichever packet happened to trigger the flush re-creates the stall that +// was just cleared. +TEST_F(BufferedPacketQueueTest, FlushAdvancesPastEverythingItDelivered) +{ + for (uint16_t s = 1; s <= 5; ++s) feed(s); + + // 6 never arrives, and 11 is missing from the run so the flush is triggered by 10 while + // 12 is already buffered behind it. + feed(7); + feed(8); + feed(9); + feed(12); + feed(10); + ASSERT_EQ(delivered, (std::vector{1, 2, 3, 4, 5, 7, 8, 9, 10, 12})); + + feed(13); + ASSERT_EQ(delivered, (std::vector{1, 2, 3, 4, 5, 7, 8, 9, 10, 12, 13})) + << "13 follows the highest packet already delivered and must not be held back"; +} + +// RTP starts at a random sequence number, so a VTX rebooting mid-session shows up as a jump +// that can be more than half the sequence space - where a signed 16 bit distance reads as +// "behind us" and the queue would never move again. +TEST_F(BufferedPacketQueueTest, StreamRestartFarAheadResyncs) +{ + feed(1); + feed(2); + + feed(40000); + ASSERT_EQ(delivered, (std::vector{1, 2})) << "40000 is held back at first"; + + advance(25); + feed(40001); + feed(40002); + + ASSERT_EQ(delivered, (std::vector{1, 2, 40000, 40001, 40002})) + << "after the age bound releases 40000 the queue has to follow the new base"; +} + +// The buffer is a hash map, so the flush has to sort it - and sorting by raw value puts a +// block that straddles the wrap point in the wrong order. +TEST_F(BufferedPacketQueueTest, FlushAcrossTheWrapDeliversInOrder) +{ + feed(65532); + + // 65533 never arrives; the rest of the run crosses zero. + feed(65534); + feed(65535); + feed(0); + feed(1); + feed(2); + feed(3); + + ASSERT_EQ(delivered, (std::vector{65532, 65534, 65535, 0, 1, 2, 3})) + << "sorted by value this comes out as 0, 1, 2, 65534, 65535"; +} + // ---------- gtest boilerplate main ----------------------------------------- int main(int argc, char** argv) {