libtgvoip/controller/net/JitterBuffer.cpp

//
// 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 <math.h>

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<int64_t>(static_cast<int64_t>(pkt->timestamp) - step * minDelay);
		first = true;
		LOGI("jitter: resyncing, next timestamp = %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;i<JITTER_SLOT_COUNT;i++){
		if(slots[i].buffer!=NULL && pkt->timestamp-slots[i].timestamp<pkt->timestamp-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 && lastPutTimestamp<nextFetchTimestamp){
		LOGV("jitter: don't have timestamp %lld, buffering", (long long int)nextFetchTimestamp);
		Advance();
		return JR_BUFFERING;
	}*/

	//needBuffering=false;

	int64_t timestampToGet = nextFetchTimestamp + offset * (int32_t)step;

	int i;
	for (i = 0; i < JITTER_SLOT_COUNT; i++)
	{
		if (!slots[i].buffer.IsEmpty() && slots[i].timestamp == timestampToGet)
		{
			break;
		}
	}

	if (i < JITTER_SLOT_COUNT)
	{
		if (pkt && pkt->size < 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;
}