Teaspeak-Server/server/src/ConnectionStatistics.cpp

//
// Created by wolverindev on 11.11.17.
//

#include <misc/memtracker.h>
#include "ConnectionStatistics.h"
#include "VirtualServer.h"

using namespace std;
using namespace std::chrono;
using namespace ts;
using namespace ts::server;
using namespace ts::stats;
using namespace ts::protocol;

ConnectionStatistics::ConnectionStatistics(const shared_ptr<ConnectionStatistics>& handle, bool properties) : handle(handle) {
    memtrack::allocated<ConnectionStatistics>(this);

    if(properties) {
        this->properties = make_shared<Properties>(); //TODO load etc?
        this->properties->register_property_type<property::ConnectionProperties>();
    }

    /*
    this->properties->registerProperty("connection_packets_sent_speech", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_bytes_sent_speech", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_packets_received_speech", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_bytes_received_speech", 0, PROP_STATISTIC);

    this->properties->registerProperty("connection_packets_sent_keepalive", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_bytes_sent_keepalive", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_packets_received_keepalive", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_bytes_received_keepalive", 0, PROP_STATISTIC);

    this->properties->registerProperty("connection_packets_sent_control", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_bytes_sent_control", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_packets_received_control", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_bytes_received_control", 0, PROP_STATISTIC);

    this->properties->registerProperty("connection_packets_sent_total", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_bytes_sent_total", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_packets_received_total", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_bytes_received_total", 0, PROP_STATISTIC);

    this->properties->registerProperty("connection_bandwidth_sent_last_second_total", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_bandwidth_sent_last_minute_total", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_bandwidth_received_last_second_total", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_bandwidth_received_last_minute_total", 0, PROP_STATISTIC);

    this->properties->registerProperty("connection_filetransfer_bandwidth_sent", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_filetransfer_bandwidth_received", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_filetransfer_bytes_sent_total", 0, PROP_STATISTIC);
    this->properties->registerProperty("connection_filetransfer_bytes_received_total", 0, PROP_STATISTIC);
    */
}

ConnectionStatistics::~ConnectionStatistics() {
    memtrack::freed<ConnectionStatistics>(this);

    {
        lock_guard lock(this->history_lock_incoming);

        for(auto entry : this->history_incoming)
            if(entry->use_count.fetch_sub(1) == 1)
                delete entry;
        for(auto entry : this->history_file_incoming)
            if(entry->use_count.fetch_sub(1) == 1)
                delete entry;

        this->history_incoming.clear();
        this->history_file_incoming.clear();
    }
    {
        lock_guard lock(this->history_lock_outgoing);

        for(auto entry : this->history_outgoing)
            if(entry->use_count.fetch_sub(1) == 1)
                delete entry;
        for(auto entry : this->history_file_outgoing)
            if(entry->use_count.fetch_sub(1) == 1)
                delete entry;

        this->history_outgoing.clear();
        this->history_file_outgoing.clear();
    }
}

std::shared_ptr<Properties> ConnectionStatistics::statistics() {
    return this->properties;
}

void ConnectionStatistics::logIncomingPacket(const category::value &category, size_t size) {
    auto info_entry = new StatisticEntry{};
    info_entry->timestamp = system_clock::now();
    info_entry->size = uint16_t(size);

    this->_log_incoming_packet(info_entry, category);
}

void ConnectionStatistics::_log_incoming_packet(ts::stats::StatisticEntry *info_entry, int8_t index) {
    if(index >= 0 && index <= 3) {
        this->connection_packets_received[index] ++;
        this->connection_bytes_received[index] += info_entry->size;
    }
    this->connection_packets_received[0] ++;
    this->connection_bytes_received[0] += info_entry->size;

    if(this->_measure_bandwidths) {
        auto lock_count = info_entry->use_count++;
        assert(lock_count >= 0);
        (void) lock_count;

        lock_guard lock(this->history_lock_incoming);
        this->history_incoming.push_back(info_entry);
    }
    if(this->handle)
        this->handle->_log_incoming_packet(info_entry, index);
}

void ConnectionStatistics::logOutgoingPacket(const category::value &category, size_t size) {
    auto info_entry = new StatisticEntry{};
    info_entry->timestamp = system_clock::now();
    info_entry->size = uint16_t(size);

    this->_log_outgoing_packet(info_entry, category);
}


void ConnectionStatistics::_log_outgoing_packet(ts::stats::StatisticEntry *info_entry, int8_t index) {
    if(index >= 0 && index <= 3) {
        this->connection_packets_sent[index] ++;
        this->connection_bytes_sent[index] += info_entry->size;
    }
    this->connection_packets_sent[0] ++;
    this->connection_bytes_sent[0] += info_entry->size;

    if(this->_measure_bandwidths) {
        auto lock_count = info_entry->use_count++;
        assert(lock_count >= 0);
        (void) lock_count;

        lock_guard lock(this->history_lock_outgoing);
        this->history_outgoing.push_back(info_entry);
    }
    if(this->handle)
        this->handle->_log_outgoing_packet(info_entry, index);
}

/* file transfer */
void ConnectionStatistics::logFileTransferIn(uint64_t bytes) {
    auto info_entry = new StatisticEntry{};
    info_entry->timestamp = system_clock::now();
    info_entry->size = bytes;

    this->_log_incoming_file_packet(info_entry);
}

void ConnectionStatistics::_log_incoming_file_packet(ts::stats::StatisticEntry *info_entry) {
    this->file_bytes_received += info_entry->size;

    if(this->_measure_bandwidths) {
        auto lock_count = info_entry->use_count++;
        assert(lock_count >= 0);
        (void) lock_count;

        lock_guard lock(this->history_lock_incoming);
        this->history_file_incoming.push_back(info_entry);
    }

    if(this->handle)
        this->handle->_log_incoming_file_packet(info_entry);
}

void ConnectionStatistics::logFileTransferOut(uint64_t bytes) {
    auto info_entry = new StatisticEntry{};
    info_entry->timestamp = system_clock::now();
    info_entry->size = bytes;

    this->_log_outgoing_file_packet(info_entry);
}

void ConnectionStatistics::_log_outgoing_file_packet(ts::stats::StatisticEntry *info_entry) {
    this->file_bytes_sent += info_entry->size;

    if(this->_measure_bandwidths) {
        auto lock_count = info_entry->use_count++;
        assert(lock_count >= 0);
        (void) lock_count;

        lock_guard lock(this->history_lock_outgoing);
        this->history_file_outgoing.push_back(info_entry);
    }

    if(this->handle)
        this->handle->_log_outgoing_file_packet(info_entry);
}

void ConnectionStatistics::tick() {
    StatisticEntry* entry;
    {
        auto timeout_min = system_clock::now() - minutes(1);

        lock_guard lock(this->history_lock_incoming);

        while(!this->history_incoming.empty() && (entry = this->history_incoming[0])->timestamp < timeout_min) {
            if(entry->use_count.fetch_sub(1) == 1)
                delete entry;

            this->history_incoming.pop_front();
        }

        while(!this->history_file_incoming.empty() && (entry = this->history_file_incoming[0])->timestamp < timeout_min) {
            if(entry->use_count.fetch_sub(1) == 1)
                delete entry;

            this->history_file_incoming.pop_front();
        }
    }
    {
        auto timeout_min = system_clock::now() - minutes(1);
        lock_guard lock(this->history_lock_outgoing);

        while(!this->history_outgoing.empty() && (entry = this->history_outgoing[0])->timestamp < timeout_min) {
            if(entry->use_count.fetch_sub(1) == 1)
                delete entry;

            this->history_outgoing.pop_front();
        }

        while(!this->history_file_outgoing.empty() && (entry = this->history_file_outgoing[0])->timestamp < timeout_min) {
            if(entry->use_count.fetch_sub(1) == 1)
                delete entry;

            this->history_file_outgoing.pop_front();
        }
    }

    if(this->properties) {
        auto& _properties = *this->properties;
#define M(type, index) \
    _properties[property::CONNECTION_BYTES_SENT_ ##type] = (uint64_t) this->connection_bytes_sent[index]; \
    _properties[property::CONNECTION_PACKETS_SENT_ ##type] = (uint64_t) this->connection_packets_sent[index]; \
    _properties[property::CONNECTION_BYTES_RECEIVED_ ##type] = (uint64_t) this->connection_bytes_received[index]; \
    _properties[property::CONNECTION_PACKETS_RECEIVED_ ##type] = (uint64_t) this->connection_packets_received[index]; \

        M(TOTAL, 0);
        M(CONTROL, 1);
        M(KEEPALIVE, 2);
        M(SPEECH, 3);

        _properties[property::CONNECTION_FILETRANSFER_BYTES_RECEIVED_TOTAL] = (uint64_t) this->file_bytes_received;
        _properties[property::CONNECTION_FILETRANSFER_BYTES_SENT_TOTAL] = (uint64_t) this->file_bytes_sent;

        _properties[property::CONNECTION_FILETRANSFER_BYTES_RECEIVED_TOTAL] = (uint64_t) this->file_bytes_received;
        _properties[property::CONNECTION_FILETRANSFER_BYTES_SENT_TOTAL] = (uint64_t) this->file_bytes_sent;
    }
}

DataSummery ConnectionStatistics::dataReport() {
    DataSummery report{};
    auto minTimeout = system_clock::now() - seconds(1);


    {
        lock_guard lock(this->history_lock_incoming);

        for(const auto& elm : this->history_incoming){
            if(elm->timestamp >= minTimeout) {
                report.recv_second += elm->size;
            }

            report.recv_minute += elm->size;
        }

        for(const auto& elm : this->history_file_incoming) {
            report.file_recv += elm->size;
        }
    }

    {
        lock_guard lock(this->history_lock_outgoing);

        for(const auto& elm : this->history_outgoing){
            if(elm->timestamp >= minTimeout) {
                report.send_second += elm->size;
            }

            report.send_minute += elm->size;
        }

        for(const auto& elm : this->history_file_outgoing) {
            report.file_send += elm->size;
        }
    }

    report.recv_minute /= 60;
    report.send_minute /= 60;
    return report;
}

FullReport ConnectionStatistics::full_report() {
    FullReport report{};

    for(size_t index = 0 ; index < 4; index++) {
        report.connection_bytes_sent[index] = (uint64_t) this->connection_bytes_sent[index];
        report.connection_packets_sent[index] = (uint64_t) this->connection_packets_sent[index];
        report.connection_bytes_received[index] = (uint64_t) this->connection_bytes_received[index];
        report.connection_packets_received[index] = (uint64_t) this->connection_packets_received[index];
    }

    report.file_bytes_sent = this->file_bytes_sent;
    report.file_bytes_received = this->file_bytes_received;

    return report;
}

std::pair<uint64_t, uint64_t> ConnectionStatistics::mark_file_bytes() {
    std::pair<uint64_t, uint64_t> result;

    {
        lock_guard lock(this->history_lock_incoming);
        if(this->mark_file_bytes_received < this->file_bytes_received)
            result.second = this->file_bytes_received - this->mark_file_bytes_received;
        this->mark_file_bytes_received = (uint64_t) this->file_bytes_received;
    }

    {

        lock_guard lock(this->history_lock_outgoing);

        if(this->mark_file_bytes_sent < this->file_bytes_sent)
            result.first = this->file_bytes_sent - this->mark_file_bytes_sent;
        this->mark_file_bytes_sent = (uint64_t) this->file_bytes_sent;
    }

    return result;
}