Digital_Voice_Modem_Project
/
dvmhost
mirror of https://github.com/DVMProject/dvmhost.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

minor jitter buffer alteration to properly drop the oldest frame vs the last frame;

Browse Source
pull/126/head
Bryan Biedenkapp 4 months ago
parent 63e197fa73
commit aea826fcbf
5 changed files with 118 additions and 17 deletions
Show Stats Download Patch File Download Diff File

27
src/common/network/AdaptiveJitterBuffer.cpp
Unescape View File

@ -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

48
src/fne/network/FNEPeerConnection.cpp
Unescape View File

@ -12,6 +12,8 @@
using namespace network;
#include <chrono>
// ---------------------------------------------------------------------------
// 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<BufferedFrame*>& readyFrames)
{
std::lock_guard<std::mutex> 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<std::mutex> lock(m_jitterMutex);
uint64_t currentTime = std::chrono::duration_cast<std::chrono::microseconds>(
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<BufferedFrame*> 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;
}

12
src/fne/network/FNEPeerConnection.h
Unescape View File

@ -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<BufferedFrame*>& readyFrames);
/**
* @brief Cleans up jitter buffer for the specified stream.
* @param streamId Stream ID.

12
src/fne/network/TrafficNetwork.cpp
Unescape View File

@ -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<BufferedFrame*> 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<BufferedFrame*> 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<BufferedFrame*> 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<BufferedFrame*> 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) {

36
tests/network/AdaptiveJitterBuffer_Tests.cpp
Unescape View File

@ -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<BufferedFrame*> readyFrames;
std::vector<BufferedFrame*> timedOutFrames;
const std::string base = "base";
const std::string older = "older";
const std::string newer = "newer";
REQUIRE(buffer.processFrame(3000U, reinterpret_cast<const uint8_t*>(base.data()), (uint32_t)base.size(), readyFrames));
cleanupFrames(readyFrames);
REQUIRE(buffer.processFrame(3002U, reinterpret_cast<const uint8_t*>(older.data()), (uint32_t)older.size(), readyFrames));
REQUIRE(readyFrames.empty());
REQUIRE(buffer.processFrame(3003U, reinterpret_cast<const uint8_t*>(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);

Write Preview
Loading…
Cancel
Save

Powered by TurnKey Linux.