diff --git a/app/videonative/src/main/cpp/BufferedPacketQueue.h b/app/videonative/src/main/cpp/BufferedPacketQueue.h index 88d6eab0..891540a8 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) @@ -196,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 @@ -214,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); } } @@ -231,19 +279,27 @@ 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()); } /** - * @brief Handles buffer overflow by processing in-order packets and discarding others. + * @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 - void restartBuffering(Callback& callback, SeqType currPacketIdx) + SeqType drainBufferInOrder(Callback& callback) { // Process as many in-order buffered packets as possible processBufferedPackets(callback); @@ -262,25 +318,40 @@ 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 + [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; 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..6e803b62 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,124 @@ 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)); +} + +// 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) { 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