TeaSpeakLibrary/src/protocol/AcknowledgeManager.cpp

#include "AcknowledgeManager.h"
#include <cmath>
#include <misc/endianness.h>

using namespace ts;
using namespace ts::connection;
using namespace ts::protocol;
using namespace std;
using namespace std::chrono;

AcknowledgeManager::AcknowledgeManager() = default;

AcknowledgeManager::~AcknowledgeManager() {
    this->reset();
}

void AcknowledgeManager::reset() {
    {
        std::unique_lock lock{this->entry_lock};
        auto pending_entries = std::move(this->entries);
        lock.unlock();

        /* save because entries are not accessable anymore */
        for(const auto& entry : pending_entries)
            if(entry->acknowledge_listener)
                entry->acknowledge_listener->executionFailed("reset");
    }
}

size_t AcknowledgeManager::awaiting_acknowledge() {
    std::lock_guard lock(this->entry_lock);
    return this->entries.size();
}

void AcknowledgeManager::process_packet(ts::protocol::BasicPacket &packet) {
    if(!packet.type().requireAcknowledge()) return;

    auto entry = make_shared<Entry>();
    entry->acknowledge_listener = std::move(packet.getListener());

    entry->buffer = packet.buffer();

    entry->resend_count = 0;
    entry->resend_period = milliseconds((int) ceil(this->average_response * 3/2));
    entry->first_send = system_clock::now();
    entry->next_resend = entry->first_send + entry->resend_period;

    entry->packet_type = packet.type().type();
    entry->packet_id = packet.packetId();

    entry->acknowledged = false;
    entry->send_count = 1;
    {
        std::lock_guard lock(this->entry_lock);
        this->entries.push_front(std::move(entry));
    }
}

bool AcknowledgeManager::process_acknowledge(uint8_t packet_type, const pipes::buffer_view& payload, std::string& error) {
    if(payload.length() < 2) return false;

    PacketType target_type{packet_type == protocol::ACK_LOW ? PacketType::COMMAND_LOW : PacketType::COMMAND};
    uint16_t target_id{be2le16((char*) payload.data_ptr())};

    std::shared_ptr<Entry> entry;
    std::unique_ptr<threads::Future<bool>> ack_listener;
    {
        std::lock_guard lock{this->entry_lock};
        for(auto it = this->entries.begin(); it != this->entries.end(); it++) {
            if((*it)->packet_type == target_type && (*it)->packet_id == target_id) {
                entry = *it;
                ack_listener = std::move(entry->acknowledge_listener); /* move it out so nobody else could call it as well */

                entry->send_count--;
                if(entry->send_count == 0)
                    this->entries.erase(it);
                break;
            }
        }
    }
    if(!entry) {
        error = "Missing packet id (" + to_string(target_id) + ")";
        return false;
    }

    auto time = system_clock::now() - entry->next_resend + entry->resend_period;
    auto ms_time = duration_cast<milliseconds>(time).count();
    if(ms_time > 5) {
        this->average_response = this->average_response * .80 + ms_time * .2;
    }

    entry->acknowledged = true;
    if(ack_listener) ack_listener->executionSucceed(true);
    return true;
}

ssize_t AcknowledgeManager::execute_resend(const system_clock::time_point& now , std::chrono::system_clock::time_point &next_resend,std::deque<pipes::buffer>& buffers, string& error) {
    size_t resend_count{0};

    vector<shared_ptr<Entry>> need_resend;
    {
        bool cleanup{false};
        std::lock_guard lock{this->entry_lock};
        need_resend.resize(this->entries.size());

        for (auto &entry : this->entries) {
            if(entry->acknowledged) {
                if(entry->next_resend + entry->resend_period <= now) { // Some resends are lost. So we just drop it after time
                    entry.reset();
                    cleanup = true;
                }
            } else {
               if(entry->next_resend <= now) {
                   entry->resend_period = entry->resend_period + milliseconds{(int) ceil(this->average_response * 2)};
                   if(entry->resend_period.count() > 1000)
                       entry->resend_period = milliseconds(1000);
                   else if(entry->resend_period.count() < 25)
                       entry->resend_period = milliseconds(25);

                   entry->next_resend = now + entry->resend_period;
                   need_resend.push_back(entry);
               }
                if(next_resend > entry->next_resend)
                    next_resend = entry->next_resend;
            }
        }

        if(cleanup) {
            this->entries.erase(std::remove_if(this->entries.begin(), this->entries.end(),
                    [](const auto& entry) { return !entry; }), this->entries.end());
        }
    }

    for(const auto& packet : need_resend) {
        if(packet->resend_count > 15 && packet->first_send + seconds(15) < now) { //FIXME configurable
            error = "Failed to receive acknowledge for packet " + to_string(packet->packet_id) + " of type " + PacketTypeInfo::fromid(packet->packet_type).name();
            return -1;
        }

        resend_count++;
        packet->resend_count++;
        packet->send_count++;
        buffers.push_back(packet->buffer);
    }

    return resend_count;
}