diff --git a/src/common/network/AdaptiveJitterBuffer.cpp b/src/common/network/AdaptiveJitterBuffer.cpp index e2da0073..7e1d2172 100644 --- a/src/common/network/AdaptiveJitterBuffer.cpp +++ b/src/common/network/AdaptiveJitterBuffer.cpp @@ -122,13 +122,30 @@ bool AdaptiveJitterBuffer::processFrame(uint16_t seq, const uint8_t* data, uint3 // frame is in the future - buffer it m_reorderedFrames++; + auto existing = m_buffer.find(seq); + if (existing != m_buffer.end()) { + delete existing->second; + m_buffer.erase(existing); + m_droppedFrames++; + } + // check buffer capacity if (m_buffer.size() >= m_maxBufferSize) { - // buffer is full - drop oldest frame to make room - auto oldestIt = m_buffer.begin(); - delete oldestIt->second; - m_buffer.erase(oldestIt); - m_droppedFrames++; + // buffer is full - drop the oldest received frame to make room + auto oldestIt = m_buffer.end(); + uint64_t oldestTimestamp = UINT64_MAX; + for (auto it = m_buffer.begin(); it != m_buffer.end(); ++it) { + if (it->second != nullptr && it->second->timestamp < oldestTimestamp) { + oldestTimestamp = it->second->timestamp; + oldestIt = it; + } + } + + if (oldestIt != m_buffer.end()) { + delete oldestIt->second; + m_buffer.erase(oldestIt); + m_droppedFrames++; + } } // add frame to buffer diff --git a/src/fne/network/FNEPeerConnection.cpp b/src/fne/network/FNEPeerConnection.cpp index d20a8d70..5e865529 100644 --- a/src/fne/network/FNEPeerConnection.cpp +++ b/src/fne/network/FNEPeerConnection.cpp @@ -12,6 +12,8 @@ using namespace network; +#include + // --------------------------------------------------------------------------- // Public Class Members // --------------------------------------------------------------------------- @@ -29,6 +31,27 @@ AdaptiveJitterBuffer* FNEPeerConnection::getOrCreateJitterBuffer(uint64_t stream return m_jitterBuffers[streamId]; } +/* Processes a frame through the jitter buffer for the specified stream. */ + +bool FNEPeerConnection::processJitterFrame(uint64_t streamId, uint16_t seq, const uint8_t* data, uint32_t length, + std::vector& readyFrames) +{ + std::lock_guard lock(m_jitterMutex); + + auto it = m_jitterBuffers.find(streamId); + if (it == m_jitterBuffers.end()) { + m_jitterBuffers[streamId] = new AdaptiveJitterBuffer(m_jitterMaxSize, m_jitterMaxWait); + it = m_jitterBuffers.find(streamId); + } + + AdaptiveJitterBuffer* buffer = (it != m_jitterBuffers.end()) ? it->second : nullptr; + if (buffer == nullptr) { + return false; + } + + return buffer->processFrame(seq, data, length, readyFrames); +} + /* Cleans up jitter buffer for the specified stream. */ void FNEPeerConnection::cleanupJitterBuffer(uint64_t streamId) @@ -37,6 +60,16 @@ void FNEPeerConnection::cleanupJitterBuffer(uint64_t streamId) auto it = m_jitterBuffers.find(streamId); if (it != m_jitterBuffers.end()) { + uint64_t totalFrames = 0ULL, reorderedFrames = 0ULL, droppedFrames = 0ULL, timedOutFrames = 0ULL; + AdaptiveJitterBuffer* buffer = it->second; + buffer->getStatistics(totalFrames, reorderedFrames, droppedFrames, timedOutFrames); + + if (reorderedFrames != 0ULL || droppedFrames != 0ULL || timedOutFrames != 0ULL) { + LogInfoEx(LOG_MASTER, "PEER %u (%s) jitter stream %u stats, total = %llu, reordered = %llu, timedOut = %llu, dropped = %llu", + id(), identWithQualifier().c_str(), (uint32_t)streamId, + (uint64_t)totalFrames, (uint64_t)reorderedFrames, (uint64_t)timedOutFrames, (uint64_t)droppedFrames); + } + delete it->second; m_jitterBuffers.erase(it); } @@ -51,17 +84,24 @@ void FNEPeerConnection::checkJitterTimeouts() } std::lock_guard lock(m_jitterMutex); + uint64_t currentTime = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch()).count(); // check timeouts for all active jitter buffers for (auto& pair : m_jitterBuffers) { AdaptiveJitterBuffer* buffer = pair.second; if (buffer != nullptr) { std::vector timedOutFrames; - buffer->checkTimeouts(timedOutFrames); + buffer->checkTimeouts(timedOutFrames, currentTime); - // note: timed-out frames are handled by the calling context - // this method just ensures the buffers are checked periodically - // the frames themselves are cleaned up by the caller + for (BufferedFrame* frame : timedOutFrames) { + if (frame != nullptr) { + LogWarning(LOG_MASTER, "PEER %u (%s) jitter stream %u timeout; delivered seq %u after %lluus", + id(), identWithQualifier().c_str(), (uint32_t)pair.first, (uint32_t)frame->seq, + (uint64_t)(currentTime - frame->timestamp)); + } + } + for (BufferedFrame* frame : timedOutFrames) { delete frame; } diff --git a/src/fne/network/FNEPeerConnection.h b/src/fne/network/FNEPeerConnection.h index 602a1cfd..d1c92ab7 100644 --- a/src/fne/network/FNEPeerConnection.h +++ b/src/fne/network/FNEPeerConnection.h @@ -145,6 +145,18 @@ namespace network */ AdaptiveJitterBuffer* getOrCreateJitterBuffer(uint64_t streamId); + /** + * @brief Processes a frame through the jitter buffer for the specified stream. + * @param streamId Stream ID. + * @param seq RTP sequence number. + * @param data Frame data buffer. + * @param length Frame length. + * @param[out] readyFrames Vector of frames ready for delivery. + * @returns bool True if frame was processed successfully, otherwise false. + */ + bool processJitterFrame(uint64_t streamId, uint16_t seq, const uint8_t* data, uint32_t length, + std::vector& readyFrames); + /** * @brief Cleans up jitter buffer for the specified stream. * @param streamId Stream ID. diff --git a/src/fne/network/TrafficNetwork.cpp b/src/fne/network/TrafficNetwork.cpp index 71560421..c3803337 100644 --- a/src/fne/network/TrafficNetwork.cpp +++ b/src/fne/network/TrafficNetwork.cpp @@ -1106,10 +1106,9 @@ void TrafficNetwork::taskNetworkRx(NetPacketRequest* req) if (network->m_tagDMR != nullptr) { // check if jitter buffer is enabled for this peer if (connection->jitterBufferEnabled() && req->rtpHeader.getSequence() != RTP_END_OF_CALL_SEQ) { - AdaptiveJitterBuffer* buffer = connection->getOrCreateJitterBuffer(streamId); std::vector readyFrames; - buffer->processFrame(req->rtpHeader.getSequence(), req->buffer, req->length, readyFrames); + connection->processJitterFrame(streamId, req->rtpHeader.getSequence(), req->buffer, req->length, readyFrames); // process all frames that are now ready (in sequence order) for (BufferedFrame* frame : readyFrames) { @@ -1147,10 +1146,9 @@ void TrafficNetwork::taskNetworkRx(NetPacketRequest* req) if (network->m_tagP25 != nullptr) { // check if jitter buffer is enabled for this peer if (connection->jitterBufferEnabled() && req->rtpHeader.getSequence() != RTP_END_OF_CALL_SEQ) { - AdaptiveJitterBuffer* buffer = connection->getOrCreateJitterBuffer(streamId); std::vector readyFrames; - buffer->processFrame(req->rtpHeader.getSequence(), req->buffer, req->length, readyFrames); + connection->processJitterFrame(streamId, req->rtpHeader.getSequence(), req->buffer, req->length, readyFrames); // process all frames that are now ready (in sequence order) for (BufferedFrame* frame : readyFrames) { @@ -1188,10 +1186,9 @@ void TrafficNetwork::taskNetworkRx(NetPacketRequest* req) if (network->m_tagNXDN != nullptr) { // check if jitter buffer is enabled for this peer if (connection->jitterBufferEnabled() && req->rtpHeader.getSequence() != RTP_END_OF_CALL_SEQ) { - AdaptiveJitterBuffer* buffer = connection->getOrCreateJitterBuffer(streamId); std::vector readyFrames; - buffer->processFrame(req->rtpHeader.getSequence(), req->buffer, req->length, readyFrames); + connection->processJitterFrame(streamId, req->rtpHeader.getSequence(), req->buffer, req->length, readyFrames); // process all frames that are now ready (in sequence order) for (BufferedFrame* frame : readyFrames) { @@ -1229,10 +1226,9 @@ void TrafficNetwork::taskNetworkRx(NetPacketRequest* req) if (network->m_tagAnalog != nullptr) { // check if jitter buffer is enabled for this peer if (connection->jitterBufferEnabled() && req->rtpHeader.getSequence() != RTP_END_OF_CALL_SEQ) { - AdaptiveJitterBuffer* buffer = connection->getOrCreateJitterBuffer(streamId); std::vector readyFrames; - buffer->processFrame(req->rtpHeader.getSequence(), req->buffer, req->length, readyFrames); + connection->processJitterFrame(streamId, req->rtpHeader.getSequence(), req->buffer, req->length, readyFrames); // process all frames that are now ready (in sequence order) for (BufferedFrame* frame : readyFrames) { diff --git a/tests/network/AdaptiveJitterBuffer_Tests.cpp b/tests/network/AdaptiveJitterBuffer_Tests.cpp index 65a638c6..a29ce6ae 100644 --- a/tests/network/AdaptiveJitterBuffer_Tests.cpp +++ b/tests/network/AdaptiveJitterBuffer_Tests.cpp @@ -124,6 +124,42 @@ TEST_CASE("AdaptiveJitterBuffer reorders and drops duplicates", "[network][jitte REQUIRE(timedOutFrames == 0U); } +TEST_CASE("AdaptiveJitterBuffer drops the oldest buffered frame on overflow", "[network][jitter]") +{ + AdaptiveJitterBuffer buffer(1U, 1000U); + std::vector readyFrames; + std::vector timedOutFrames; + + const std::string base = "base"; + const std::string older = "older"; + const std::string newer = "newer"; + + REQUIRE(buffer.processFrame(3000U, reinterpret_cast(base.data()), (uint32_t)base.size(), readyFrames)); + cleanupFrames(readyFrames); + + REQUIRE(buffer.processFrame(3002U, reinterpret_cast(older.data()), (uint32_t)older.size(), readyFrames)); + REQUIRE(readyFrames.empty()); + + REQUIRE(buffer.processFrame(3003U, reinterpret_cast(newer.data()), (uint32_t)newer.size(), readyFrames)); + REQUIRE(readyFrames.empty()); + + buffer.checkTimeouts(timedOutFrames, UINT64_MAX); + REQUIRE(timedOutFrames.size() == 1U); + REQUIRE(timedOutFrames[0U]->seq == 3003U); + REQUIRE(frameToString(timedOutFrames[0U]) == newer); + cleanupFrames(timedOutFrames); + + uint64_t totalFrames = 0ULL; + uint64_t reorderedFrames = 0ULL; + uint64_t droppedFrames = 0ULL; + uint64_t timedOutFramesCount = 0ULL; + buffer.getStatistics(totalFrames, reorderedFrames, droppedFrames, timedOutFramesCount); + REQUIRE(totalFrames == 3U); + REQUIRE(reorderedFrames == 2U); + REQUIRE(droppedFrames == 1U); + REQUIRE(timedOutFramesCount == 1U); +} + TEST_CASE("AdaptiveJitterBuffer handles wraparound and timeout delivery", "[network][jitter]") { AdaptiveJitterBuffer buffer(4U, 1000U);