// // libtgvoip is free and unencumbered public domain software. // For more information, see http://unlicense.org or the UNLICENSE file // you should have received with this source code distribution. // #include "VoIPController.h" #include "controller/net/JitterBuffer.h" #include "tools/logging.h" #include "VoIPServerConfig.h" #include using namespace tgvoip; JitterBuffer::JitterBuffer(uint32_t step) : step(step), slots{} { if (step < 30) { minMinDelay = ServerConfig::GetSharedInstance()->GetUInt("jitter_min_delay_20", 6); maxMinDelay = ServerConfig::GetSharedInstance()->GetUInt("jitter_max_delay_20", 25); maxUsedSlots = ServerConfig::GetSharedInstance()->GetUInt("jitter_max_slots_20", 50); } else if (step < 50) { minMinDelay = ServerConfig::GetSharedInstance()->GetUInt("jitter_min_delay_40", 4); maxMinDelay = ServerConfig::GetSharedInstance()->GetUInt("jitter_max_delay_40", 15); maxUsedSlots = ServerConfig::GetSharedInstance()->GetUInt("jitter_max_slots_40", 30); } else { minMinDelay = ServerConfig::GetSharedInstance()->GetUInt("jitter_min_delay_60", 2); maxMinDelay = ServerConfig::GetSharedInstance()->GetUInt("jitter_max_delay_60", 10); maxUsedSlots = ServerConfig::GetSharedInstance()->GetUInt("jitter_max_slots_60", 20); } lossesToReset = ServerConfig::GetSharedInstance()->GetUInt("jitter_losses_to_reset", 20); resyncThreshold = ServerConfig::GetSharedInstance()->GetDouble("jitter_resync_threshold", 1.0); #ifdef TGVOIP_DUMP_JITTER_STATS #ifdef TGVOIP_JITTER_DUMP_FILE dump = fopen(TGVOIP_JITTER_DUMP_FILE, "w"); #elif defined(__ANDROID__) dump = fopen("/sdcard/tgvoip_jitter_dump.txt", "w"); #else dump = fopen("tgvoip_jitter_dump.txt", "w"); #endif tgvoip_log_file_write_header(dump); fprintf(dump, "PTS\tRTS\tNumInBuf\tAJitter\tADelay\tTDelay\n"); #endif Reset(); } JitterBuffer::~JitterBuffer() { Reset(); } void JitterBuffer::SetMinPacketCount(uint32_t count) { LOGI("jitter: set min packet count %u", count); minDelay = count; minMinDelay = count; //Reset(); } int JitterBuffer::GetMinPacketCount() { return (int)minDelay; } double JitterBuffer::GetTimeoutWindow() { return (lossesToReset * step) / 1000.0; } void JitterBuffer::HandleInput(unsigned char *data, size_t len, uint32_t timestamp, bool isEC) { MutexGuard m(mutex); jitter_packet_t pkt; pkt.size = len; pkt.buffer = Buffer::Wrap(data, len, [](void *) {}, [](void *a, size_t) -> void * { return a; }); pkt.timestamp = timestamp; pkt.isEC = isEC; PutInternal(&pkt, !isEC); //LOGV("in, ts=%d, ec=%d", timestamp, isEC); } void JitterBuffer::PutInternal(jitter_packet_t *pkt, bool overwriteExisting) { if (pkt->size > JITTER_SLOT_SIZE) { LOGE("The packet is too big to fit into the jitter buffer"); return; } int i; for (i = 0; i < JITTER_SLOT_COUNT; i++) { if (!slots[i].buffer.IsEmpty() && slots[i].timestamp == pkt->timestamp) { //LOGV("Found existing packet for timestamp %u, overwrite %d", pkt->timestamp, overwriteExisting); if (overwriteExisting) { slots[i].buffer.CopyFrom(pkt->buffer, pkt->size); slots[i].size = pkt->size; slots[i].isEC = pkt->isEC; } return; } } gotSinceReset++; if (wasReset) { wasReset = false; outstandingDelayChange = 0; nextFetchTimestamp = static_cast(static_cast(pkt->timestamp) - step * minDelay); first = true; LOGI("jitter: resyncing, next timeDecodeNextFramestamp = %lld (step=%d, minDelay=%f)", (long long int)nextFetchTimestamp, step, minDelay); } for (i = 0; i < JITTER_SLOT_COUNT; i++) { if (!slots[i].buffer.IsEmpty()) { // Clear packets older than the last fetched packet if (slots[i].timestamp < nextFetchTimestamp - 1) { slots[i].buffer = Buffer(); } } } /*double prevTime=0; uint32_t closestTime=0; for(i=0;itimestamp-slots[i].timestamptimestamp-closestTime){ closestTime=slots[i].timestamp; prevTime=slots[i].recvTime; } }*/ // Time deviation check double time = VoIPController::GetCurrentTime(); if (expectNextAtTime != 0) { double dev = expectNextAtTime - time; //LOGV("packet dev %f", dev); deviationHistory.Add(dev); expectNextAtTime += step / 1000.0; } else { expectNextAtTime = time + step / 1000.0; } // Late packet check if (pkt->timestamp < nextFetchTimestamp) { //LOGW("jitter: would drop packet with timestamp %d because it is late but not hopelessly", pkt->timestamp); latePacketCount++; lostPackets--; } else if (pkt->timestamp < nextFetchTimestamp - 1) { //LOGW("jitter: dropping packet with timestamp %d because it is too late", pkt->timestamp); latePacketCount++; return; } if (pkt->timestamp > lastPutTimestamp) lastPutTimestamp = pkt->timestamp; for (i = 0; i < JITTER_SLOT_COUNT; i++) { if (slots[i].buffer.IsEmpty()) break; } if (i == JITTER_SLOT_COUNT || GetCurrentDelay() >= maxUsedSlots) { int toRemove = JITTER_SLOT_COUNT; uint32_t oldestTimestamp = 0xFFFFFFFF; for (i = 0; i < JITTER_SLOT_COUNT; i++) { if (!slots[i].buffer.IsEmpty() && slots[i].timestamp < oldestTimestamp) { toRemove = i; oldestTimestamp = slots[i].timestamp; } } Advance(); slots[toRemove].buffer = Buffer(); i = toRemove; } slots[i].timestamp = pkt->timestamp; slots[i].size = pkt->size; slots[i].buffer = bufferPool.Get(); slots[i].recvTimeDiff = time - prevRecvTime; slots[i].isEC = pkt->isEC; slots[i].buffer.CopyFrom(pkt->buffer, pkt->size); #ifdef TGVOIP_DUMP_JITTER_STATS fprintf(dump, "%u\t%.03f\t%d\t%.03f\t%.03f\t%.03f\n", pkt->timestamp, time, GetCurrentDelay(), lastMeasuredJitter, lastMeasuredDelay, minDelay); #endif prevRecvTime = time; } void JitterBuffer::Reset() { wasReset = true; needBuffering = true; lastPutTimestamp = 0; int i; for (i = 0; i < JITTER_SLOT_COUNT; i++) { if (!slots[i].buffer.IsEmpty()) { slots[i].buffer = Buffer(); } } delayHistory.Reset(); lateHistory.Reset(); adjustingDelay = false; lostSinceReset = 0; gotSinceReset = 0; expectNextAtTime = 0; deviationHistory.Reset(); outstandingDelayChange = 0; dontChangeDelay = 0; } size_t JitterBuffer::HandleOutput(unsigned char *buffer, size_t len, int offsetInSteps, bool advance, int &playbackScaledDuration, bool &isEC) { jitter_packet_t pkt; pkt.buffer = Buffer::Wrap(buffer, len, [](void *) {}, [](void *a, size_t) -> void * { return a; }); pkt.size = len; MutexGuard m(mutex); if (first) { first = false; unsigned int delay = GetCurrentDelay(); if (GetCurrentDelay() > 5) { LOGW("jitter: delay too big upon start (%u), dropping packets", delay); for (; delay > GetMinPacketCount(); --delay) { for (int i = 0; i < JITTER_SLOT_COUNT; i++) { if (slots[i].timestamp == nextFetchTimestamp) { if (!slots[i].buffer.IsEmpty()) { slots[i].buffer = Buffer(); } break; } } Advance(); } } } int result = GetInternal(&pkt, offsetInSteps, advance); if (outstandingDelayChange != 0) { if (outstandingDelayChange < 0) { playbackScaledDuration = 40; outstandingDelayChange += 20; } else { playbackScaledDuration = 80; outstandingDelayChange -= 20; } //LOGV("outstanding delay change: %d", outstandingDelayChange); } else if (advance && GetCurrentDelay() == 0) { //LOGV("stretching packet because the next one is late"); playbackScaledDuration = 80; } else { playbackScaledDuration = 60; } if (result == JR_OK) { isEC = pkt.isEC; return pkt.size; } else { return 0; } } int JitterBuffer::GetInternal(jitter_packet_t *pkt, int offset, bool advance) { /*if(needBuffering && lastPutTimestampsize < slots[i].size) { LOGE("jitter: packet won't fit into provided buffer of %d (need %d)", int(slots[i].size), int(pkt->size)); } else { if (pkt) { pkt->size = slots[i].size; pkt->timestamp = slots[i].timestamp; pkt->buffer.CopyFrom(slots[i].buffer, slots[i].size); pkt->isEC = slots[i].isEC; } } slots[i].buffer = Buffer(); if (offset == 0) Advance(); lostCount = 0; needBuffering = false; return JR_OK; } LOGV("jitter: found no packet for timestamp %lld (last put = %d, lost = %d)", (long long int)timestampToGet, lastPutTimestamp, lostCount); if (advance) Advance(); if (!needBuffering) { lostCount++; if (offset == 0) { lostPackets++; lostSinceReset++; } if (lostCount >= lossesToReset || (gotSinceReset > minDelay * 25 && lostSinceReset > gotSinceReset / 2)) { LOGW("jitter: lost %d packets in a row, resetting", lostCount); //minDelay++; dontIncMinDelay = 16; dontDecMinDelay += 128; if (GetCurrentDelay() < minDelay) nextFetchTimestamp -= (int64_t)(minDelay - GetCurrentDelay()); lostCount = 0; Reset(); } return JR_MISSING; } return JR_BUFFERING; } void JitterBuffer::Advance() { nextFetchTimestamp += step; } unsigned int JitterBuffer::GetCurrentDelay() { unsigned int delay = 0; int i; for (i = 0; i < JITTER_SLOT_COUNT; i++) { if (!slots[i].buffer.IsEmpty()) delay++; } return delay; } void JitterBuffer::Tick() { MutexGuard m(mutex); int i; lateHistory.Add(latePacketCount); latePacketCount = 0; bool absolutelyNoLatePackets = lateHistory.Max() == 0; double avgLate16 = lateHistory.Average(16); //LOGV("jitter: avg late=%.1f, %.1f, %.1f", avgLate16, avgLate32, avgLate64); if (avgLate16 >= resyncThreshold) { LOGV("resyncing: avgLate16=%f, resyncThreshold=%f", avgLate16, resyncThreshold); wasReset = true; } if (absolutelyNoLatePackets) { if (dontDecMinDelay > 0) dontDecMinDelay--; } delayHistory.Add(GetCurrentDelay()); avgDelay = delayHistory.Average(32); double stddev = 0; double avgdev = deviationHistory.Average(); for (i = 0; i < 64; i++) { double d = (deviationHistory[i] - avgdev); stddev += (d * d); } stddev = sqrt(stddev / 64); uint32_t stddevDelay = std::clamp((uint32_t)ceil(stddev * 2 * 1000 / step), minMinDelay, maxMinDelay); if (stddevDelay != minDelay) { int32_t diff = std::clamp((int32_t)(stddevDelay - minDelay), -1, 1); if (diff > 0) { dontDecMinDelay = 100; } if ((diff > 0 && dontIncMinDelay == 0) || (diff < 0 && dontDecMinDelay == 0)) { //nextFetchTimestamp+=diff*(int32_t)step; minDelay += diff; outstandingDelayChange += diff * 60; dontChangeDelay += 32; //LOGD("new delay from stddev %f", minDelay); if (diff < 0) { dontDecMinDelay += 25; } if (diff > 0) { dontIncMinDelay = 25; } } } lastMeasuredJitter = stddev; lastMeasuredDelay = stddevDelay; //LOGV("stddev=%.3f, avg=%.3f, ndelay=%d, dontDec=%u", stddev, avgdev, stddevDelay, dontDecMinDelay); if (dontChangeDelay) { --dontChangeDelay; } else { if (avgDelay > minDelay + 0.5) { outstandingDelayChange -= avgDelay > minDelay + 2 ? 60 : 20; dontChangeDelay += 10; } else if (avgDelay < minDelay - 0.3) { outstandingDelayChange += 20; dontChangeDelay += 10; } } //LOGV("jitter: avg delay=%d, delay=%d, late16=%.1f, dontDecMinDelay=%d", avgDelay, delayHistory[0], avgLate16, dontDecMinDelay); /*if(!adjustingDelay) { if (((minDelay==1 ? (avgDelay>=3) : (avgDelay>=minDelay/2)) && delayHistory[0]>minDelay && avgLate16<=0.1 && absolutelyNoLatePackets && dontDecMinDelay<32 && min>minDelay)) { LOGI("jitter: need adjust"); adjustingDelay=true; } }else{ if(!absolutelyNoLatePackets){ LOGI("jitter: done adjusting because we're losing packets"); adjustingDelay=false; }else if(tickCount%5==0){ LOGD("jitter: removing a packet to reduce delay"); GetInternal(NULL, 0); expectNextAtTime=0; if(GetCurrentDelay()<=minDelay || min<=minDelay){ adjustingDelay = false; LOGI("jitter: done adjusting"); } } }*/ tickCount++; } void JitterBuffer::GetAverageLateCount(double *out) { out[0] = lateHistory.Average(16); out[1] = lateHistory.Average(32); out[2] = lateHistory.Average(); } int JitterBuffer::GetAndResetLostPacketCount() { MutexGuard m(mutex); int r = lostPackets; lostPackets = 0; return r; } double JitterBuffer::GetLastMeasuredJitter() { return lastMeasuredJitter; } double JitterBuffer::GetLastMeasuredDelay() { return lastMeasuredDelay; } double JitterBuffer::GetAverageDelay() { return avgDelay; }