congestvpn/VpnPeer.cpp

#include "UdpVpn.hpp"
#include "congestion_control.hpp"

#include <stdint.h>
#include <string.h>

const double RTTLogger::EXP_AVG_FACTOR = 0.75;
const unsigned int RTTLogger::BASE_UPDATE_DELAY = 1000; // ms

VpnPeer::VpnPeer(UdpVpn* vpn, const sockaddr_in6& ext_addr,
        const in6_addr& int_addr)
    : _vpn(vpn), _ext_addr(ext_addr), _int_addr(int_addr), _next_send_seqno(0),
    _tot_bytes_sent(0), _prev_tot_bytes_sent(0),
    _congestion_controller(*this)
{
    clock_gettime(CLOCK_MONOTONIC, &_prev_tick_time);
    cycle_next_control();
}

void VpnPeer::make_loss_report() {
    VpnTlvLossReport report = VpnTlvLossReport::create(*_next_control_packet);
    report.set_report_seqno(_packet_loss.get_cur_seqno());
    report.set_losses(_packet_loss.get_tot_losses());
}

void VpnPeer::cycle_next_control() {
    _next_control_packet =
        std::make_unique<VpnControlPacket>(_vpn->get_mtu(), false);
    _next_control_packet->set_peer(this);
}

void VpnPeer::set_int_addr(const in6_addr& int_addr) {
    memcpy(&_int_addr, &int_addr, sizeof(_int_addr));
}

void VpnPeer::tick() {
    struct timespec cur_time;
    clock_gettime(CLOCK_MONOTONIC, &cur_time);

    // Compute byte rate
    _outbound_byte_rate = /* byte per second */
        (double)(_tot_bytes_sent - _prev_tot_bytes_sent)
        / timespec_us_diff(cur_time, _prev_tick_time)
        * 1e6;

    _prev_tot_bytes_sent = _tot_bytes_sent;
    _prev_tick_time = cur_time;
}

void VpnPeer::log_loss_report(const VpnTlvLossReport& loss_rep) {
    _loss_reports.prev_seqno = _loss_reports.last_seqno;
    _loss_reports.prev_losses = _loss_reports.last_losses;

    _loss_reports.last_seqno = loss_rep.get_report_seqno();
    _loss_reports.last_losses = loss_rep.get_losses();

    _congestion_controller.update_lossbased();
}

size_t VpnPeer::write(const char* data, size_t len) {
    ssize_t nsent;

    nsent = sendto(_vpn->get_socket_fd(), data, len, MSG_CONFIRM,
            (const struct sockaddr*) &_ext_addr, sizeof(_ext_addr));
    if(nsent < 0)
        throw NetError("Could not send UDP packet", errno, true);

    _tot_bytes_sent += nsent;

    return (size_t) nsent;
}

size_t VpnPeer::write(const VpnPacket& packet) {
    return write(packet.get_data(), packet.get_data_size());
}

void VpnPeer::got_inbound_packet(const VpnPacket& packet) {
    _packet_loss.log_packet(packet.get_seqno());
}

bool VpnPeer::send_control_packet() {
    if(_rtt.update_due(!_next_control_packet->is_empty())) {
        VpnTlvRTTQ::create(*_next_control_packet);
        make_loss_report();
    }

    if(!_next_control_packet->is_empty()) {
        _next_control_packet->prepare_for_sending();
        _vpn->transmit_to_peer(*_next_control_packet);
        cycle_next_control();
        return true;
    }
    return false;
}

void VpnPeer::make_rtta_for(const VpnTlvRTTQ& rttq) {
    VpnTlvRTTA rtta = VpnTlvRTTA::create(*_next_control_packet);
    rtta.set_exp_ts(rttq.get_packet().get_sending_timestamp());
    rtta.set_recv_ts(rttq.get_packet().get_reception_timestamp());
}

PacketLossLogger::PacketLossLogger() :
    _cur_seqno(0), _tot_losses(0), _last_window_losses(0), _win_start_losses(0)
{}

void PacketLossLogger::log_packet(uint32_t seqno) {
    int64_t diff = (int64_t)seqno - _cur_seqno;

    if(diff == 1) {
        _cur_seqno++;
        maybe_start_window();
        while(_received_ahead.test((_cur_seqno + 1) % PACKET_LOST_AFTER)) {
            _cur_seqno++;
            _received_ahead.reset(_cur_seqno % PACKET_LOST_AFTER);
            maybe_start_window();
        }
    } else if(LIKELY(diff > 1)) {
        if(diff < PACKET_LOST_AFTER)
            _received_ahead.set(seqno % PACKET_LOST_AFTER);
        else if(diff < PACKET_LOSS_WINDOW) {
            // Packet too much forwards -- consider _cur_seqno lost
            if(_cur_seqno % PACKET_LOSS_WINDOW > seqno % PACKET_LOSS_WINDOW) {
                // This loss crosses a window border
                for(int offs=0; offs < PACKET_LOST_AFTER; ++offs) {
                    maybe_start_window(offs);
                    if(!_received_ahead[(_cur_seqno + offs) % PACKET_LOST_AFTER])
                        _tot_losses++;
                }
            } else {
                _tot_losses += PACKET_LOST_AFTER - _received_ahead.count();
            }
            _received_ahead.reset();
            _cur_seqno = seqno;
        } else
            reboot(); // This is a huge gap -- reboot
    } else {
        if(diff < - 2*PACKET_LOSS_WINDOW)
            reboot(); // this is too much backwards -- something's wrong, reboot
        // else: ignore, we've moved forward and counted the packet as lost
    }
}

void PacketLossLogger::reboot() {
    _received_ahead.reset();
    // _tot_losses unchanged
}

void PacketLossLogger::maybe_start_window(int offs) {
    if(_cur_seqno + offs % PACKET_LOSS_WINDOW == 0) {
        _last_window_losses = _win_start_losses;
        _win_start_losses = _tot_losses;
    }
}

RTTLogger::RTTLogger() : _avg_rtt(0), _cur_rtt(0) {
    clock_gettime(CLOCK_MONOTONIC, &_last_update);
    _last_update.tv_sec--;
    // This avoids situations where both peers try to renew their RTT at the
    // same time.
    _update_delay = BASE_UPDATE_DELAY + rand() % 100 - 50;
}

void RTTLogger::log(const VpnTlvRTTA& rtt_answer) {
    uint32_t local_t0 = rtt_answer.get_exp_ts(),
             remote_t0 = rtt_answer.get_recv_ts(),
             remote_t1 = rtt_answer.get_packet().get_sending_timestamp(),
             local_t1 = rtt_answer.get_packet().get_reception_timestamp();

    _cur_rtt = (local_t1 - local_t0) - (remote_t1 - remote_t0);
    if(UNLIKELY(_avg_rtt == 0))
        _avg_rtt = _cur_rtt;
    else
        _avg_rtt = (1 - EXP_AVG_FACTOR) * _avg_rtt
            + EXP_AVG_FACTOR * _cur_rtt;

    clock_gettime(CLOCK_MONOTONIC, &_last_update);
}

bool RTTLogger::update_due(bool soon) const {
    unsigned int ms_since_last_update =
        timespec_us_ellapsed(_last_update) / 1000;
    return ms_since_last_update >= (soon ? _update_delay/2 : _update_delay);
}