danog/libtgvoip
1
0
Fork 0
mirror of https://github.com/danog/libtgvoip.git synced 2024-11-26 12:14:39 +01:00
Code Issues Projects Releases Wiki Activity

Re-enable tweaked congestion control

Browse Source
This commit is contained in:
Daniil Gentili 2020-03-30 19:10:20 +02:00
parent 2252eb1a20
commit 9ff55f6deb
Signed by: danog
GPG Key ID: 8C1BE3B34B230CA7
10 changed files with 385 additions and 271 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

45
README.md
View File

@ -1,3 +1,48 @@
# Daniil Gentili's submission to the VoIP contest 3
In this submission, I created a new and improved protocol that unifies all flags, lengths that were previously sparsed around the protocol into one single data payload.
This allows saving around 6 bytes for each data packet, and enormously reduces complexity of existing code, splitting business logic and (legacy) presentation/transport details into separate classes.
All signaling packets (including init and initAck) are now unified into extra signaling fields.
A TL-B scheme for the new protocol can be viewed in `schemeNew.tlb` (compared to the old protocol in `scheme.tlb`).
Constructor serialization and parsing is implemented using a manually-written polymorphic approach, inspired by Telegram's C++ applications (tdesktop, tdlib).
My implementation of polymorphic TL serialization and parsing can be viewed in `controller/protocol/protocol/*` and `controller/protocol/packets/PacketStructs.*`.
**The new libtgvoip is 100% backwards compatibile with previous versions, keeping compatibility even with layer 8 and below, while enormously reducing complexity, encapsulating all legacy behaviour implementation details into a single `Legacy.cpp` file.**
Among other features, all packet data buffers are moved around as `unique_ptr`s instead of being copied: previously, feeding a packet to the OPUS decoder through the jitter required **two** copies, and allocation of a slot in a buffer pool; now, a single `unique_ptr` (generated when initially parsing the packet) is moved around, instead (`protocol/Protocol.cpp`).
Similarly efficient move-logic is used for outgoing audio buffers (EC & main), network socket payloads, and reliable packet resending (see below).
The new protocol also allows for (backwards-compatible) transport multiplexing for streams, simply by making use of the packet seq (in newer protocols) or the audio PTS (in older protocols), without using additional fields.
This allows `AudioPacketSender.cpp` to reliably determine whether a certain packet was received by the other end or not, and resend only the missing audio pieces, especially in the case of Error Correction (EC) packets generated by the secondary low-bitrate opus encoder: only packets that are known to be lost (as of now) are re-sent, by using a Mask array type instead of a basic Array, skipping audio pieces that were already acknowledged for the current stream by the other endpoint.
I've also kept the reliable packet resending for the main audio stream, by further building onto and refactoring the reliable packet logic, which was now merged with the extraSignaling reliability logic (`Reliable.cpp`).
I've fixed a few nasty jitter buffer bugs:
- Fixed a bug where the jitter buffer would wrongly reset in presence of EC packets (5ba6e99882bdc045ae64a3b4f65039910677edf0; for latest commit JitterBuffer.cpp line 142-151, wrongly interpreted EC packets as late main packets)
- Fixed a minor bug where the jitter buffer would backtrack to an invalid timestamp when trying to fill gaps in transmission (457576e93faf8e99c49f936685d5c1582ca221bc; for latest commit JitterBuffer.cpp line 314, delay not multiplied by OPUS step)
Among other changes, the data structure representing incoming and outgoing streams was split into several polymorphic data structures and classes (`protocol/packets/PacketStructs.h`, `protocol/Stream.h`) to save memory and properly separate business logic from datastructures.
The commit history will be available @ https://gitlab.com/danog/libtgvoip as usual.
Please note that commit 2252eb1a2056dc63ec09e4f1acc1c7bfa6fec9e4 contains an initial final version that suitably fixes one specific issues with audio quality: in case of sudden but short packet loss bursts (like when switching from WiFi to mobile network), libtgvoip would **usually** enable redundant EC, and most importantly **decrease the bitrate of both main OPUS stream to the absolute minimum**.
This causes hearable and persistent audio quality issues even if packet loss is reduced right afterwards (since it takes a while for the congestion controller to raise the audio quality again).
In that commit, I **completely disable manual OPUS bandwidth regulation**, instead relying on EC packets to fill in the gap during the short period of downtime: this yielded **excellent results** in my tests (I used a real-life device, switching constantly from wifi to weak 4g, with only very brief switches to EC hearable in the audio stream, automatically switching back to the high-quality 20kbps stream right afterwards).
The same results were confirmed by the `tgvoip-test-suite`: after re-enabling congestion control in the **next commit**, albeit with some fixes to account for packet resending: this also yielded good results even in low-bandwidth conditions.
If I have time until the end of the submissions, I'll refactor the jitter buffer to avoid excessive switching between the main and EC audio streams in case of packet loss in low-bandwidth conditions.
In the meantime, just wanted to say that I really enjoyed working on this contest: as usual, I have many more ideas for improvements, but for now, I'm pretty satisfied of the result.
Thank you.
---
# Daniil Gentili's submission to the VoIP contest 2
My submission consists of a major refactor (to C++) of the existing libtgvoip library.

5
controller/audio/AudioPacketSender.h
View File

@ -41,6 +41,11 @@ public:
return extraEcLevel;
}
inline const double getResendCount() const
{
return resendCount;
}
inline void setShittyInternetMode(bool shittyInternetMode)
{
this->shittyInternetMode = shittyInternetMode;

488
controller/audio/OpusDecoder.cpp
View File

@ -7,10 +7,10 @@
#include "controller/audio/OpusDecoder.h"
#include "audio/Resampler.h"
#include "tools/logging.h"
#include <algorithm>
#include <cassert>
#include <cmath>
#include <cstdlib>
#include <algorithm>
#if defined HAVE_CONFIG_H || defined TGVOIP_USE_INSTALLED_OPUS
#include <opus/opus.h>
@ -25,317 +25,319 @@ using namespace tgvoip;
tgvoip::OpusDecoder::OpusDecoder(const std::shared_ptr<MediaStreamItf> &dst, bool isAsync, bool needEC)
{
dst->SetCallback(OpusDecoder::Callback, this);
Initialize(isAsync, needEC);
dst->SetCallback(OpusDecoder::Callback, this);
Initialize(isAsync, needEC);
}
tgvoip::OpusDecoder::OpusDecoder(const std::unique_ptr<MediaStreamItf> &dst, bool isAsync, bool needEC)
{
dst->SetCallback(OpusDecoder::Callback, this);
Initialize(isAsync, needEC);
dst->SetCallback(OpusDecoder::Callback, this);
Initialize(isAsync, needEC);
}
void tgvoip::OpusDecoder::Initialize(bool isAsync, bool needEC)
{
async = isAsync;
if (async)
{
decodedQueue = new BlockingQueue<Buffer>(33);
semaphore = new Semaphore(32, 0);
}
else
{
decodedQueue = NULL;
semaphore = NULL;
}
dec = opus_decoder_create(48000, 1, NULL);
if (needEC)
ecDec = opus_decoder_create(48000, 1, NULL);
else
ecDec = NULL;
async = isAsync;
if (async)
{
decodedQueue = new BlockingQueue<Buffer>(33);
semaphore = new Semaphore(32, 0);
}
else
{
decodedQueue = NULL;
semaphore = NULL;
}
dec = opus_decoder_create(48000, 1, NULL);
if (needEC)
ecDec = opus_decoder_create(48000, 1, NULL);
else
ecDec = NULL;
#ifdef ANDROID
buffer = reinterpret_cast<unsigned char*>(std::malloc(8192));
buffer = reinterpret_cast<unsigned char *>(std::malloc(8192));
#else
buffer = reinterpret_cast<unsigned char*>(std::aligned_alloc(2, 8192));
buffer = reinterpret_cast<unsigned char *>(std::aligned_alloc(2, 8192));
#endif
lastDecoded = NULL;
outputBufferSize = 0;
echoCanceller = NULL;
frameDuration = 20;
consecutiveLostPackets = 0;
enableDTX = false;
silentPacketCount = 0;
levelMeter = NULL;
nextLen = 0;
running = false;
remainingDataLen = 0;
processedBuffer = NULL;
prevWasEC = false;
prevLastSample = 0;
lastDecoded = NULL;
outputBufferSize = 0;
echoCanceller = NULL;
frameDuration = 20;
consecutiveLostPackets = 0;
enableDTX = false;
silentPacketCount = 0;
levelMeter = NULL;
nextLen = 0;
running = false;
remainingDataLen = 0;
processedBuffer = NULL;
prevWasEC = false;
prevLastSample = 0;
}
tgvoip::OpusDecoder::~OpusDecoder()
{
opus_decoder_destroy(dec);
if (ecDec)
opus_decoder_destroy(ecDec);
free(buffer);
if (decodedQueue)
delete decodedQueue;
if (semaphore)
delete semaphore;
opus_decoder_destroy(dec);
if (ecDec)
opus_decoder_destroy(ecDec);
free(buffer);
if (decodedQueue)
delete decodedQueue;
if (semaphore)
delete semaphore;
}
void tgvoip::OpusDecoder::SetEchoCanceller(const std::shared_ptr<EchoCanceller> &canceller)
{
echoCanceller = canceller;
echoCanceller = canceller;
}
size_t tgvoip::OpusDecoder::Callback(unsigned char *data, size_t len, void *param)
{
return reinterpret_cast<OpusDecoder *>(param)->HandleCallback(data, len);
return reinterpret_cast<OpusDecoder *>(param)->HandleCallback(data, len);
}
size_t tgvoip::OpusDecoder::HandleCallback(unsigned char *data, size_t len)
{
if (async)
{
if (!running)
{
memset(data, 0, len);
return 0;
}
if (outputBufferSize == 0)
{
outputBufferSize = len;
int packetsNeeded;
if (len > PACKET_SIZE)
packetsNeeded = len / PACKET_SIZE;
else
packetsNeeded = 1;
packetsNeeded *= 2;
semaphore->Release(packetsNeeded);
}
assert(outputBufferSize == len && "output buffer size is supposed to be the same throughout callbacks");
if (len == PACKET_SIZE)
{
Buffer lastDecoded = decodedQueue->GetBlocking();
if (lastDecoded.IsEmpty())
return 0;
memcpy(data, *lastDecoded, PACKET_SIZE);
semaphore->Release();
if (silentPacketCount > 0)
{
silentPacketCount--;
if (levelMeter)
levelMeter->Update(reinterpret_cast<int16_t *>(data), 0);
return 0;
}
if (echoCanceller)
{
echoCanceller->SpeakerOutCallback(data, PACKET_SIZE);
}
}
else
{
LOGE("Opus decoder buffer length != 960 samples");
abort();
}
}
else
{
if (remainingDataLen == 0 && silentPacketCount == 0)
{
int duration = DecodeNextFrame();
remainingDataLen = (size_t)(duration / 20 * 960 * 2);
}
if (silentPacketCount > 0 || remainingDataLen == 0 || !processedBuffer)
{
if (silentPacketCount > 0)
silentPacketCount--;
memset(data, 0, 960 * 2);
if (levelMeter)
levelMeter->Update(reinterpret_cast<int16_t *>(data), 0);
return 0;
}
memcpy(data, processedBuffer, 960 * 2);
remainingDataLen -= 960 * 2;
if (remainingDataLen > 0)
{
memmove(processedBuffer, processedBuffer + 960 * 2, remainingDataLen);
}
}
if (levelMeter)
levelMeter->Update(reinterpret_cast<int16_t *>(data), len / 2);
return len;
if (async)
{
if (!running)
{
memset(data, 0, len);
return 0;
}
if (outputBufferSize == 0)
{
outputBufferSize = len;
int packetsNeeded;
if (len > PACKET_SIZE)
packetsNeeded = len / PACKET_SIZE;
else
packetsNeeded = 1;
packetsNeeded *= 2;
semaphore->Release(packetsNeeded);
}
assert(outputBufferSize == len && "output buffer size is supposed to be the same throughout callbacks");
if (len == PACKET_SIZE)
{
Buffer lastDecoded = decodedQueue->GetBlocking();
if (lastDecoded.IsEmpty())
return 0;
memcpy(data, *lastDecoded, PACKET_SIZE);
semaphore->Release();
if (silentPacketCount > 0)
{
silentPacketCount--;
if (levelMeter)
levelMeter->Update(reinterpret_cast<int16_t *>(data), 0);
return 0;
}
if (echoCanceller)
{
echoCanceller->SpeakerOutCallback(data, PACKET_SIZE);
}
}
else
{
LOGE("Opus decoder buffer length != 960 samples");
abort();
}
}
else
{
if (remainingDataLen == 0 && silentPacketCount == 0)
{
int duration = DecodeNextFrame();
remainingDataLen = (size_t)(duration / 20 * 960 * 2);
}
if (silentPacketCount > 0 || remainingDataLen == 0 || !processedBuffer)
{
if (silentPacketCount > 0)
silentPacketCount--;
memset(data, 0, 960 * 2);
if (levelMeter)
levelMeter->Update(reinterpret_cast<int16_t *>(data), 0);
return 0;
}
memcpy(data, processedBuffer, 960 * 2);
remainingDataLen -= 960 * 2;
if (remainingDataLen > 0)
{
memmove(processedBuffer, processedBuffer + 960 * 2, remainingDataLen);
}
}
if (levelMeter)
levelMeter->Update(reinterpret_cast<int16_t *>(data), len / 2);
return len;
}
void tgvoip::OpusDecoder::Start()
{
if (!async)
return;
running = true;
thread = new Thread(std::bind(&tgvoip::OpusDecoder::RunThread, this));
thread->SetName("opus_decoder");
thread->SetMaxPriority();
thread->Start();
if (!async)
return;
running = true;
thread = new Thread(std::bind(&tgvoip::OpusDecoder::RunThread, this));
thread->SetName("opus_decoder");
thread->SetMaxPriority();
thread->Start();
}
void tgvoip::OpusDecoder::Stop()
{
if (!running || !async)
return;
running = false;
semaphore->Release();
thread->Join();
delete thread;
if (!running || !async)
return;
running = false;
semaphore->Release();
thread->Join();
delete thread;
}
void tgvoip::OpusDecoder::RunThread()
{
int i;
LOGI("decoder: packets per frame %d", packetsPerFrame);
while (running)
{
int playbackDuration = DecodeNextFrame();
for (i = 0; i < playbackDuration / 20; i++)
{
semaphore->Acquire();
if (!running)
{
LOGI("==== decoder exiting ====");
return;
}
try
{
Buffer buf = bufferPool.Get();
if (remainingDataLen > 0)
{
for (auto &effect : postProcEffects)
{
effect->Process(reinterpret_cast<int16_t *>(processedBuffer + (PACKET_SIZE * i)), 960);
}
buf.CopyFrom(processedBuffer + (PACKET_SIZE * i), 0, PACKET_SIZE);
}
else
{
//LOGE("Error decoding, result=%d", size);
memset(*buf, 0, PACKET_SIZE);
}
decodedQueue->Put(std::move(buf));
}
catch (std::bad_alloc &x)
{
LOGW("decoder: no buffers left!");
}
}
}
int i;
LOGI("decoder: packets per frame %d", packetsPerFrame);
while (running)
{
int playbackDuration = DecodeNextFrame();
for (i = 0; i < playbackDuration / 20; i++)
{
semaphore->Acquire();
if (!running)
{
LOGI("==== decoder exiting ====");
return;
}
try
{
Buffer buf = bufferPool.Get();
if (remainingDataLen > 0)
{
for (auto &effect : postProcEffects)
{
effect->Process(reinterpret_cast<int16_t *>(processedBuffer + (PACKET_SIZE * i)), 960);
}
buf.CopyFrom(processedBuffer + (PACKET_SIZE * i), 0, PACKET_SIZE);
}
else
{
//LOGE("Error decoding, result=%d", size);
memset(*buf, 0, PACKET_SIZE);
}
decodedQueue->Put(std::move(buf));
}
catch (std::bad_alloc &x)
{
LOGW("decoder: no buffers left!");
}
}
}
}
int tgvoip::OpusDecoder::DecodeNextFrame()
{
int playbackDuration = 0;
bool isEC = false;
auto ptr = jitterBuffer->HandleOutput(true, playbackDuration, isEC);
bool fec = false;
if (!ptr)
{
fec = true;
ptr = jitterBuffer->HandleOutput( false, playbackDuration, isEC);
/*if (len)
int playbackDuration = 0;
bool isEC = false;
auto ptr = jitterBuffer->HandleOutput(true, playbackDuration, isEC);
bool fec = false;
if (!ptr)
{
fec = true;
ptr = jitterBuffer->HandleOutput(false, playbackDuration, isEC);
/*if (len)
LOGV("Trying FEC...");*/
}
int size;
if (ptr)
{
size = opus_decode(isEC ? ecDec : dec, **ptr, ptr->Length(), (opus_int16 *)decodeBuffer, packetsPerFrame * 960, fec ? 1 : 0);
consecutiveLostPackets = 0;
if (prevWasEC != isEC && size)
{
LOGE("========== JITTER: Switching from %s to %s ==========", prevWasEC ? "EC" : "main", isEC ? "EC" : "main");
// It turns out the waveforms generated by the PLC feature are also great to help smooth out the
// otherwise audible transition between the frames from different decoders. Those are basically an extrapolation
// of the previous successfully decoded data -- which is exactly what we need here.
size = opus_decode(prevWasEC ? ecDec : dec, NULL, 0, reinterpret_cast<opus_int16*>(nextBuffer), packetsPerFrame * 960, 0);
if (size) {
int16_t* plcSamples = reinterpret_cast<int16_t*>(nextBuffer);
int16_t* samples = reinterpret_cast<int16_t*>(decodeBuffer);
}
int size;
if (ptr)
{
size = opus_decode(isEC ? ecDec : dec, **ptr, ptr->Length(), (opus_int16 *)decodeBuffer, packetsPerFrame * 960, fec ? 1 : 0);
consecutiveLostPackets = 0;
if (prevWasEC != isEC && size)
{
LOGE("========== JITTER: Switching from %s to %s ==========", prevWasEC ? "EC" : "main", isEC ? "EC" : "main");
// It turns out the waveforms generated by the PLC feature are also great to help smooth out the
// otherwise audible transition between the frames from different decoders. Those are basically an extrapolation
// of the previous successfully decoded data -- which is exactly what we need here.
size = opus_decode(prevWasEC ? ecDec : dec, NULL, 0, reinterpret_cast<opus_int16 *>(nextBuffer), packetsPerFrame * 960, 0);
if (size)
{
int16_t *plcSamples = reinterpret_cast<int16_t *>(nextBuffer);
int16_t *samples = reinterpret_cast<int16_t *>(decodeBuffer);
constexpr float coeffs[] = {0.999802f, 0.995062f, 0.984031f, 0.966778f, 0.943413f, 0.914084f, 0.878975f, 0.838309f, 0.792344f, 0.741368f,
0.685706f, 0.625708f, 0.561754f, 0.494249f, 0.423619f, 0.350311f, 0.274788f, 0.197527f, 0.119018f, 0.039757f};
// But why 20 samples? Typically the sample size is 60....
for (int i = 0; i < 20; i++) {
// But why 20 samples? Typically the sample size is 60....
for (int i = 0; i < 20; i++)
{
samples[i] = static_cast<int16_t>(round(plcSamples[i] * coeffs[i] + samples[i] * (1.f - coeffs[i])));
}
}
}
prevWasEC = isEC;
prevLastSample = decodeBuffer[size - 1];
}
else
{ // do packet loss concealment
if (++consecutiveLostPackets > 2 && enableDTX)
{
silentPacketCount += packetsPerFrame;
size = packetsPerFrame * 960;
}
else
{
size = opus_decode(prevWasEC ? ecDec : dec, NULL, 0, (opus_int16 *)decodeBuffer, packetsPerFrame * 960, 0);
LOGE("========== JITTER: OPUS PLC ==========");
//LOGV("PLC");
}
}
if (size < 0)
LOGW("decoder: opus_decode error %d", size);
remainingDataLen = size;
if (playbackDuration == 80)
{
//LOGW("Rescaling to 80");
processedBuffer = buffer;
audio::Resampler::Rescale60To80((int16_t *)decodeBuffer, (int16_t *)processedBuffer);
}
else if (playbackDuration == 40)
{
//LOGW("Rescaling to 40");
processedBuffer = buffer;
audio::Resampler::Rescale60To40((int16_t *)decodeBuffer, (int16_t *)processedBuffer);
}
else
{
processedBuffer = decodeBuffer;
}
return playbackDuration;
}
}
}
prevWasEC = isEC;
prevLastSample = decodeBuffer[size - 1];
}
else
{ // do packet loss concealment
if (++consecutiveLostPackets > 2 && enableDTX)
{
silentPacketCount += packetsPerFrame;
size = packetsPerFrame * 960;
}
else
{
size = opus_decode(prevWasEC ? ecDec : dec, NULL, 0, (opus_int16 *)decodeBuffer, packetsPerFrame * 960, 0);
LOGE("========== JITTER: OPUS PLC ==========");
//LOGV("PLC");
}
}
if (size < 0)
LOGW("decoder: opus_decode error %d", size);
remainingDataLen = size;
if (playbackDuration == 80)
{
//LOGW("Rescaling to 80");
processedBuffer = buffer;
audio::Resampler::Rescale60To80((int16_t *)decodeBuffer, (int16_t *)processedBuffer);
}
else if (playbackDuration == 40)
{
//LOGW("Rescaling to 40");
processedBuffer = buffer;
audio::Resampler::Rescale60To40((int16_t *)decodeBuffer, (int16_t *)processedBuffer);
}
else
{
processedBuffer = decodeBuffer;
}
return playbackDuration;
}
void tgvoip::OpusDecoder::SetFrameDuration(uint32_t duration)
{
frameDuration = duration;
packetsPerFrame = frameDuration / 20;
frameDuration = duration;
packetsPerFrame = frameDuration / 20;
}
void tgvoip::OpusDecoder::SetJitterBuffer(const std::shared_ptr<JitterBuffer> &jitterBuffer)
{
this->jitterBuffer = jitterBuffer;
this->jitterBuffer = jitterBuffer;
}
void tgvoip::OpusDecoder::SetDTX(bool enable)
{
enableDTX = enable;
enableDTX = enable;
}
void tgvoip::OpusDecoder::SetLevelMeter(const std::shared_ptr<AudioLevelMeter> &levelMeter)
{
this->levelMeter = levelMeter;
this->levelMeter = levelMeter;
}
void tgvoip::OpusDecoder::AddAudioEffect(const std::shared_ptr<effects::AudioEffect> &effect)
{
postProcEffects.push_back(effect);
postProcEffects.push_back(effect);
}
void tgvoip::OpusDecoder::RemoveAudioEffect(const std::shared_ptr<effects::AudioEffect> &effect)
{
auto i = std::find(postProcEffects.begin(), postProcEffects.end(), effect);
if (i != postProcEffects.end())
postProcEffects.erase(i);
auto i = std::find(postProcEffects.begin(), postProcEffects.end(), effect);
if (i != postProcEffects.end())
postProcEffects.erase(i);
}

6
controller/audio/OpusEncoder.cpp
View File

@ -47,8 +47,9 @@ tgvoip::OpusEncoder::OpusEncoder(const std::shared_ptr<MediaStreamItf> &source,
opus_encoder_ctl(enc, OPUS_SET_INBAND_FEC(1));
opus_encoder_ctl(enc, OPUS_SET_SIGNAL(OPUS_SIGNAL_VOICE));
opus_encoder_ctl(enc, OPUS_SET_BANDWIDTH(OPUS_AUTO));
opus_encoder_ctl(enc, OPUS_SET_BITRATE(OPUS_AUTO));
requestedBitrate = 20000;
currentBitrate = 0;
currentBitrate = 20000;
running = false;
echoCanceller = NULL;
complexity = 10;
@ -106,7 +107,7 @@ void tgvoip::OpusEncoder::Stop()
void tgvoip::OpusEncoder::SetBitrate(uint32_t bitrate)
{
LOGE("=============== SETTING BITRATE TO %u ===============", bitrate);
requestedBitrate = 20000;
requestedBitrate = bitrate;
}
void tgvoip::OpusEncoder::Encode(int16_t *data, size_t len)
@ -117,6 +118,7 @@ void tgvoip::OpusEncoder::Encode(int16_t *data, size_t len)
currentBitrate = requestedBitrate;
LOGV("opus_encoder: setting bitrate to %u", currentBitrate);
}
if (levelMeter)
levelMeter->Update(data, len);
if (secondaryEncoderEnabled != wasSecondaryEncoderEnabled)

2
controller/controller/Init.cpp
View File

@ -34,6 +34,8 @@ VoIPController::VoIPController() : rawSendQueue(64)
maxUnsentStreamPackets = ServerConfig::GetSharedInstance()->GetUInt("max_unsent_stream_packets", 2);
unackNopThreshold = ServerConfig::GetSharedInstance()->GetUInt("unack_nop_threshold", 10);
initAudioBitrate = 20000;
//audioBitrateStepDecr /= 2;
auto aStm = std::make_shared<OutgoingAudioStream>();
aStm->codec = Codec::Opus;

56
controller/net/CongestionControl.cpp
View File

@ -18,9 +18,7 @@ using namespace tgvoip;
CongestionControlPacket::CongestionControlPacket(uint32_t _seq, uint8_t _streamId) : seq(_seq), streamId(_streamId){};
CongestionControlPacket::CongestionControlPacket(const Packet &pkt) : seq(pkt.seq), streamId(pkt.streamId){};
CongestionControl::CongestionControl() : cwnd(static_cast<size_t>(ServerConfig::GetSharedInstance()->GetInt("audio_congestion_window", 1024))),
max(cwnd * 1.1),
min(cwnd * 0.9)
CongestionControl::CongestionControl() : cwnd(static_cast<size_t>(ServerConfig::GetSharedInstance()->GetInt("audio_congestion_window", 1024)))
{
}
@ -145,23 +143,59 @@ void CongestionControl::Tick()
inflightHistory.Add(inflightDataSize);
}
int CongestionControl::GetBandwidthControlAction()
CongestionControl::Action CongestionControl::GetBandwidthControlAction(int netMode, double multiply)
{
if (VoIPController::GetCurrentTime() - lastActionTime < 1)
return TGVOIP_CONCTL_ACT_NONE;
return None;
Action action;
size_t inflightAvg = GetInflightDataSize();
size_t max = (cwnd * multiply) * 1.1;
size_t min = (cwnd * multiply) * 0.9;
LOGW("inflightAvg=%lu, max=%lu, min=%lu", inflightAvg, max, min);
if (inflightAvg < min)
{
lastActionTime = VoIPController::GetCurrentTime();
return TGVOIP_CONCTL_ACT_INCREASE;
action = Increase;
}
if (inflightAvg > max)
else if (inflightAvg > max)
{
lastActionTime = VoIPController::GetCurrentTime();
return TGVOIP_CONCTL_ACT_DECREASE;
action = Decrease;
}
return TGVOIP_CONCTL_ACT_NONE;
else
{
action = None;
}
uint8_t actAfter = 3;
if (netMode < NET_TYPE_LTE)
{
actAfter--;
}
if (netMode <= NET_TYPE_GPRS)
{
actAfter--;
}
if (netMode <= NET_TYPE_EDGE)
{
actAfter--;
}
if (action == lastAction)
{
LOGE("Would act on %hhu, current tries %hhu, act after %hhu", action, lastActionCount, actAfter);
if (lastActionCount++ == actAfter)
{
lastActionTime = VoIPController::GetCurrentTime();
lastActionCount = 0;
return action;
}
}
else
{
lastActionCount = 0;
LOGE("Would act on %hhu, current tries %hhu, act after %hhu", action, lastActionCount, actAfter);
}
lastAction = action;
return None;
}
uint32_t CongestionControl::GetSendLossCount()

20
controller/net/CongestionControl.h
View File

@ -13,11 +13,8 @@
#include <map>
#include <stdlib.h>
#define TGVOIP_CONCTL_ACT_INCREASE 1
#define TGVOIP_CONCTL_ACT_DECREASE 2
#define TGVOIP_CONCTL_ACT_NONE 0
#define TGVOIP_CONCTL_LOST_AFTER 2
#define TGVOIP_CONCTL_ACT_AFTER 2
namespace tgvoip
{
@ -47,6 +44,14 @@ public:
CongestionControl();
~CongestionControl();
enum Action : uint8_t
{
None = 0,
Increase = 1,
Decrease = 2,
Min = 3
};
void PacketSent(const CongestionControlPacket &pkt, size_t size);
void PacketLost(const CongestionControlPacket &pkt);
void PacketAcknowledged(const CongestionControlPacket &pkt);
@ -57,14 +62,17 @@ public:
size_t GetCongestionWindow();
size_t GetAcknowledgedDataSize();
void Tick();
int GetBandwidthControlAction();
Action GetBandwidthControlAction(int netMode, double multiply = 1.0);
uint32_t GetSendLossCount();
private:
HistoricBuffer<double, 100> rttHistory;
HistoricBuffer<size_t, 30> inflightHistory;
std::array<tgvoip_congestionctl_packet_t, 100> inflightPackets{};
uint32_t lossCount = 0;
Action lastAction = None;
uint8_t lastActionCount = 0;
double tmpRtt = 0.0;
double lastActionTime = 0;
double lastActionRtt = 0;
@ -75,8 +83,6 @@ private:
size_t inflightDataSize = 0;
size_t cwnd;
size_t max;
size_t min;
};
} // namespace tgvoip

3
controller/net/JitterBuffer.h
View File

@ -70,7 +70,8 @@ private:
//BufferPool<JITTER_SLOT_SIZE, JITTER_SLOT_COUNT> bufferPool;
Mutex mutex;
uint32_t step;
int64_t lastMain = 0;
std::array<jitter_packet_t, JITTER_SLOT_COUNT> slots;
std::atomic<int64_t> nextFetchTimestamp{0}; // What frame to read next (protected for GetSeqTooLate)
std::atomic<double> minDelay{6};

5
controller/protocol/Bandwidth.cpp
View File

@ -2,7 +2,6 @@
using namespace tgvoip;
#pragma mark - Bandwidth management
double VoIPController::GetAverageRTT()
@ -117,21 +116,25 @@ void VoIPController::UpdateAudioBitrateLimit()
if (dataSavingMode || dataSavingRequestedByPeer)
{
maxBitrate = maxAudioBitrateSaving;
LOGE("=============== SETTING BITRATE FROM BANDWIDTH CONTROLLER: saving %u ===============", initAudioBitrateSaving);
encoder->SetBitrate(initAudioBitrateSaving);
}
else if (networkType == NET_TYPE_GPRS)
{
maxBitrate = maxAudioBitrateGPRS;
LOGE("=============== SETTING BITRATE FROM BANDWIDTH CONTROLLER: GPRS %u ===============", initAudioBitrateGPRS);
encoder->SetBitrate(initAudioBitrateGPRS);
}
else if (networkType == NET_TYPE_EDGE)
{
maxBitrate = maxAudioBitrateEDGE;
LOGE("=============== SETTING BITRATE FROM BANDWIDTH CONTROLLER: EDGE %u ===============", initAudioBitrateEDGE);
encoder->SetBitrate(initAudioBitrateEDGE);
}
else
{
maxBitrate = maxAudioBitrate;
LOGE("=============== SETTING BITRATE FROM BANDWIDTH CONTROLLER: NORMAL %u ===============", initAudioBitrate);
encoder->SetBitrate(initAudioBitrate);
}
encoder->SetVadMode(dataSavingMode || dataSavingRequestedByPeer);

26
controller/protocol/Tick.cpp
View File

@ -3,7 +3,6 @@
using namespace tgvoip;
void VoIPController::SetState(int state)
{
this->state = state;
@ -276,24 +275,39 @@ void VoIPController::UpdateAudioBitrate()
SetState(STATE_FAILED);
}
int act = conctl.GetBandwidthControlAction();
if (dynamic_cast<AudioPacketSender *>(GetStreamByType<OutgoingAudioStream>()->packetSender.get())->getShittyInternetMode())
auto *sender = dynamic_cast<AudioPacketSender *>(GetStreamByType<OutgoingAudioStream>()->packetSender.get());
double multiple = 1.0;
double resendCount = sender->getResendCount();
uint8_t ecLevel = sender->getExtraEcLevel();
if (ecLevel)
{
multiple *= ecLevel / 2;
}
if (resendCount)
{
multiple *= resendCount;
}
auto act = conctl.GetBandwidthControlAction(sender->getShittyInternetMode(), multiple);
if (act == CongestionControl::Min)
{
//encoder->SetBitrate(8000);
}
else if (act == TGVOIP_CONCTL_ACT_DECREASE)
else if (act == CongestionControl::Decrease)
{
LOGE("==== DECREASING BITRATE ======");
LOGE("=============== DECREASING BITRATE ===============");
uint32_t bitrate = encoder->GetBitrate();
if (bitrate > 8000)
encoder->SetBitrate(bitrate < (minAudioBitrate + audioBitrateStepDecr) ? minAudioBitrate : (bitrate - audioBitrateStepDecr));
}
else if (act == TGVOIP_CONCTL_ACT_INCREASE)
else if (act == CongestionControl::Increase)
{
LOGE("=============== INCREASING BITRATE ===============");
uint32_t bitrate = encoder->GetBitrate();
if (bitrate < maxBitrate)
encoder->SetBitrate(bitrate + audioBitrateStepIncr);
}
LOGE("========= CURRENT BITRATE=%u =========", encoder->GetBitrate())
if (state == STATE_ESTABLISHED && time - lastRecvPacketTime >= reconnectingTimeout)
{