Compare commits

..

6 Commits

Author SHA1 Message Date
08a083f346 fix missing include for gcc12 2023-10-15 16:31:58 +02:00
2a6dca69c7 fix msvc conversion error 2023-06-16 17:15:40 +02:00
27aeb03a61 make max (packet) segment size configurable 2023-06-16 02:04:00 +02:00
304aae05c2 ledbat tweaks 2023-03-12 18:48:33 +01:00
95bfa2473c implement dynamic timeout 2023-03-12 01:24:11 +01:00
dfeb569aee minor optional stuff 2023-01-31 23:51:15 +01:00
4 changed files with 81 additions and 76 deletions

View File

@ -5,75 +5,67 @@
#include <cmath> #include <cmath>
#include <deque> #include <deque>
#include <cstdint> #include <cstdint>
#include <tuple>
#include <cassert> #include <cassert>
#include <iomanip> #include <iomanip>
#include <iostream> #include <iostream>
#include <limits> #include <limits>
LEDBAT::LEDBAT(void) { // https://youtu.be/0HRwNSA-JYM
_time_start_offset = clock::now();
{ // add some high delay values inline constexpr bool PLOTTING = false;
// spec want +inf
//_rtt_buffer.push_back(_base_delay); LEDBAT::LEDBAT(size_t maximum_segment_data_size) : MAXIMUM_SEGMENT_DATA_SIZE(maximum_segment_data_size) {
//_rtt_buffer.push_back(_base_delay); _time_start_offset = clock::now();
//_rtt_buffer.push_back(_base_delay);
}
} }
size_t LEDBAT::canSend(void) const { size_t LEDBAT::canSend(void) const {
if (_in_flight.empty()) { if (_in_flight.empty()) {
return 496u; return MAXIMUM_SEGMENT_DATA_SIZE;
} }
//const float time_since_last_sent {std::min(
//getTimeNow() - std::get<1>(_in_flight.back()),
//0.01f // 10ms max
//)};
//const float bps {std::min(
//(_cwnd / getCurrentDelay()),
//max_byterate_allowed
//)};
const int64_t cspace = _cwnd - _in_flight_bytes; const int64_t cspace = _cwnd - _in_flight_bytes;
if (cspace < 496) { if (cspace < MAXIMUM_SEGMENT_DATA_SIZE) {
return 0u; return 0u;
} }
const int64_t fspace = _fwnd - _in_flight_bytes; const int64_t fspace = _fwnd - _in_flight_bytes;
if (fspace < 496) { if (fspace < MAXIMUM_SEGMENT_DATA_SIZE) {
return 0u; return 0u;
} }
size_t space = std::ceil(std::min(cspace, fspace) / 496.f) * 496.f; size_t space = std::ceil(std::min<float>(cspace, fspace) / MAXIMUM_SEGMENT_DATA_SIZE) * MAXIMUM_SEGMENT_DATA_SIZE;
// data size, no overhead
//const int64_t can_send_size {std::min<int64_t>(
//bps * time_since_last_sent - segment_overhead,
//maximum_segment_size - segment_overhead
//)};
//const int64_t can_send_size {static_cast<int64_t>(bps * time_since_last_sent - segment_overhead)};
//if (can_send_size < 100) {
//return 0;
//} else {
//return can_send_size;
//}
return space; return space;
} }
std::vector<LEDBAT::SeqIDType> LEDBAT::getTimeouts(void) const {
std::vector<LEDBAT::SeqIDType> list;
// after 2 delays we trigger timeout
const auto now_adjusted = getTimeNow() - getCurrentDelay()*2.f;
for (const auto& [seq, time_stamp, size] : _in_flight) {
if (now_adjusted > time_stamp) {
list.push_back(seq);
}
}
return list;
}
void LEDBAT::onSent(SeqIDType seq, size_t data_size) { void LEDBAT::onSent(SeqIDType seq, size_t data_size) {
if (true) { if (true) {
for (const auto& it : _in_flight) { for (const auto& it : _in_flight) {
assert(std::get<0>(it) != seq); assert(std::get<0>(it) != seq);
} }
} }
_in_flight.push_back({seq, getTimeNow(), data_size + segment_overhead});
_in_flight_bytes += data_size + segment_overhead; _in_flight.push_back({seq, getTimeNow(), data_size + SEGMENT_OVERHEAD});
_recently_sent_bytes += data_size + segment_overhead; _in_flight_bytes += data_size + SEGMENT_OVERHEAD;
_recently_sent_bytes += data_size + SEGMENT_OVERHEAD;
} }
void LEDBAT::onAck(std::vector<SeqIDType> seqs) { void LEDBAT::onAck(std::vector<SeqIDType> seqs) {
@ -98,7 +90,7 @@ void LEDBAT::onAck(std::vector<SeqIDType> seqs) {
most_recent = std::max(most_recent, std::get<1>(*it)); most_recent = std::max(most_recent, std::get<1>(*it));
_in_flight_bytes -= std::get<2>(*it); _in_flight_bytes -= std::get<2>(*it);
_recently_acked_data += std::get<2>(*it); _recently_acked_data += std::get<2>(*it);
assert(_in_flight_bytes >= 0); assert(_in_flight_bytes >= 0); // TODO: this triggers
_in_flight.erase(it); _in_flight.erase(it);
} }
} }
@ -107,16 +99,12 @@ void LEDBAT::onAck(std::vector<SeqIDType> seqs) {
return; // not found, ignore return; // not found, ignore
} }
//addRTT(now - most_recent);
updateWindows(); updateWindows();
// update cto - no? we dont handle timeouts
} }
void LEDBAT::onLoss(SeqIDType seq, bool discard) { void LEDBAT::onLoss(SeqIDType seq, bool discard) {
auto it = std::find_if(_in_flight.begin(), _in_flight.end(), [seq](const auto& v) -> bool { auto it = std::find_if(_in_flight.begin(), _in_flight.end(), [seq](const auto& v) -> bool {
assert(!std::isnan(std::get<1>(v)));
return std::get<0>(v) == seq; return std::get<0>(v) == seq;
}); });
@ -129,7 +117,7 @@ void LEDBAT::onLoss(SeqIDType seq, bool discard) {
// at most once per rtt? // at most once per rtt?
if (false) { if (PLOTTING) {
std::cerr << "CCA: onLoss: TIME: " << getTimeNow() << "\n"; std::cerr << "CCA: onLoss: TIME: " << getTimeNow() << "\n";
} }
@ -137,7 +125,9 @@ void LEDBAT::onLoss(SeqIDType seq, bool discard) {
if (discard) { if (discard) {
_in_flight_bytes -= std::get<2>(*it); _in_flight_bytes -= std::get<2>(*it);
assert(_in_flight_bytes >= 0); assert(_in_flight_bytes >= 0);
_in_flight.erase(it);
} }
// TODO: reset timestamp?
updateWindows(); updateWindows();
} }
@ -202,19 +192,19 @@ void LEDBAT::updateWindows(void) {
if (now - _last_cwnd >= current_delay) { if (now - _last_cwnd >= current_delay) {
const float queuing_delay {current_delay - _base_delay}; const float queuing_delay {current_delay - _base_delay};
_fwnd = max_byterate_allowed * getCurrentDelay(); _fwnd = max_byterate_allowed * current_delay;
_fwnd *= 1.3f; // try do balance conservative algo a bit, current_delay _fwnd *= 1.3f; // try do balance conservative algo a bit, current_delay
//const float gain {1}; // TODO: move and increase
float gain {1.f / std::min(16.f, std::ceil(2.f*target_delay/_base_delay))}; float gain {1.f / std::min(16.f, std::ceil(2.f*target_delay/_base_delay))};
//gain *= 400.f; // from packets to bytes ~ //gain *= 400.f; // from packets to bytes ~
gain *= _recently_acked_data/10.f; // from packets to bytes ~ gain *= _recently_acked_data/5.f; // from packets to bytes ~
//gain *= 0.1f; //gain *= 0.1f;
if (_recently_lost_data) { if (_recently_lost_data) {
_cwnd = std::clamp( _cwnd = std::clamp(
_cwnd / 2.f, _cwnd / 2.f,
2.f * maximum_segment_size, //_cwnd / 1.6f,
2.f * MAXIMUM_SEGMENT_SIZE,
_cwnd _cwnd
); );
} else { } else {
@ -222,28 +212,27 @@ void LEDBAT::updateWindows(void) {
// "Multiplicative decrease" // "Multiplicative decrease"
const float constant {2.f}; // spec recs 1 const float constant {2.f}; // spec recs 1
if (queuing_delay < target_delay) { if (queuing_delay < target_delay) {
_cwnd += gain;
_cwnd = std::min( _cwnd = std::min(
_cwnd + gain, _cwnd + gain,
_fwnd _fwnd
); );
} else if (queuing_delay > target_delay) { } else if (queuing_delay > target_delay) {
_cwnd = std::clamp( _cwnd = std::clamp(
_cwnd + std::max( // TODO: where to put bytes_newly_acked _cwnd + std::max(
gain - constant * _cwnd * (queuing_delay / target_delay - 1.f), gain - constant * _cwnd * (queuing_delay / target_delay - 1.f),
-_cwnd/2.f // at most halve -_cwnd/2.f // at most halve
), ),
// never drop below 2 "packets" in flight // never drop below 2 "packets" in flight
//2.f * maximum_segment_size, 2.f * MAXIMUM_SEGMENT_SIZE,
2.f * 496,
current_delay * max_byterate_allowed // cap rate // cap rate
_fwnd
); );
} // no else, we on point. very unlikely with float } // no else, we on point. very unlikely with float
} }
if (false) { // plotting if (PLOTTING) { // plotting
std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " cwnd: " << _cwnd << "\n"; std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " cwnd: " << _cwnd << "\n";
std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " fwnd: " << _fwnd << "\n"; std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " fwnd: " << _fwnd << "\n";
std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " current_delay: " << current_delay << "\n"; std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " current_delay: " << current_delay << "\n";

View File

@ -18,31 +18,35 @@ struct LEDBAT {
static constexpr size_t UDP_HEADER_SIZE {8}; static constexpr size_t UDP_HEADER_SIZE {8};
// TODO: tcp AND IPv6 will be different // TODO: tcp AND IPv6 will be different
static constexpr size_t segment_overhead { static constexpr size_t SEGMENT_OVERHEAD {
4+ // ft overhead 4+ // ft overhead
46+ // tox? 46+ // tox?
UDP_HEADER_SIZE+ UDP_HEADER_SIZE+
IPV4_HEADER_SIZE IPV4_HEADER_SIZE
}; };
static constexpr size_t maximum_segment_size {496 + segment_overhead}; // tox 500 - 4 from ft // TODO: make configurable, set with tox ngc lossy packet size
static_assert(maximum_segment_size == 574); // mesured in wireshark //const size_t MAXIMUM_SEGMENT_DATA_SIZE {1000-4};
const size_t MAXIMUM_SEGMENT_DATA_SIZE {500-4};
//static constexpr size_t maximum_segment_size {496 + segment_overhead}; // tox 500 - 4 from ft
const size_t MAXIMUM_SEGMENT_SIZE {MAXIMUM_SEGMENT_DATA_SIZE + SEGMENT_OVERHEAD}; // tox 500 - 4 from ft
//static_assert(maximum_segment_size == 574); // mesured in wireshark
// ledbat++ says 60ms, we might need other values if relayed // ledbat++ says 60ms, we might need other values if relayed
const float target_delay {0.060f}; //const float target_delay {0.060f};
//const float target_delay {0.030f}; const float target_delay {0.030f};
//const float target_delay {0.120f}; // 2x if relayed? //const float target_delay {0.120f}; // 2x if relayed?
// TODO: use a factor for multiple of rtt // TODO: use a factor for multiple of rtt
static constexpr size_t current_delay_filter_window {16*4*2}; static constexpr size_t current_delay_filter_window {16*4};
//static constexpr size_t rtt_buffer_size_max {2000}; //static constexpr size_t rtt_buffer_size_max {2000};
float max_byterate_allowed {10*1024*1024}; // 10MiB/s float max_byterate_allowed {10*1024*1024}; // 10MiB/s
//float max_byterate_allowed {2*1024*1024};
public: public:
LEDBAT(void); LEDBAT(size_t maximum_segment_data_size);
// return the current believed window in bytes of how much data can be inflight, // return the current believed window in bytes of how much data can be inflight,
// without overstepping the delay requirement // without overstepping the delay requirement
@ -55,6 +59,9 @@ struct LEDBAT {
// respect max_byterate_allowed // respect max_byterate_allowed
size_t canSend(void) const; size_t canSend(void) const;
// get the list of timed out seq_ids
std::vector<SeqIDType> getTimeouts(void) const;
public: // callbacks public: // callbacks
// data size is without overhead // data size is without overhead
void onSent(SeqIDType seq, size_t data_size); void onSent(SeqIDType seq, size_t data_size);
@ -84,7 +91,7 @@ struct LEDBAT {
private: // state private: // state
//float _cto {2.f}; // congestion timeout value in seconds //float _cto {2.f}; // congestion timeout value in seconds
float _cwnd {2.f * maximum_segment_size}; // in bytes float _cwnd {2.f * MAXIMUM_SEGMENT_SIZE}; // in bytes
float _base_delay {2.f}; // lowest mesured delay in _rtt_buffer in seconds float _base_delay {2.f}; // lowest mesured delay in _rtt_buffer in seconds
float _last_cwnd {0.f}; // timepoint of last cwnd correction float _last_cwnd {0.f}; // timepoint of last cwnd correction

View File

@ -11,6 +11,7 @@
// TODO: should i really use both? // TODO: should i really use both?
#include <unordered_map> #include <unordered_map>
#include <map> #include <map>
#include <set>
#include <optional> #include <optional>
#include <cassert> #include <cassert>
#include <cstdio> #include <cstdio>
@ -118,7 +119,7 @@ struct NGC_FT1 {
struct Group { struct Group {
struct Peer { struct Peer {
LEDBAT cca; LEDBAT cca{500-4}; // TODO: replace with tox_group_max_custom_lossy_packet_length()-4
struct RecvTransfer { struct RecvTransfer {
uint32_t file_kind; uint32_t file_kind;
@ -216,10 +217,12 @@ void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx, float time_delta) {
for (auto& [group_number, group] : ngc_ft1_ctx->groups) { for (auto& [group_number, group] : ngc_ft1_ctx->groups) {
for (auto& [peer_number, peer] : group.peers) { for (auto& [peer_number, peer] : group.peers) {
//for (auto& tf_opt : peer.send_transfers) { auto timeouts = peer.cca.getTimeouts();
std::set<LEDBAT::SeqIDType> timeouts_set{timeouts.cbegin(), timeouts.cend()};
for (size_t idx = 0; idx < peer.send_transfers.size(); idx++) { for (size_t idx = 0; idx < peer.send_transfers.size(); idx++) {
auto& tf_opt = peer.send_transfers[idx]; auto& tf_opt = peer.send_transfers[idx];
if (tf_opt) { if (tf_opt.has_value()) {
auto& tf = tf_opt.value(); auto& tf = tf_opt.value();
tf.time_since_activity += time_delta; tf.time_since_activity += time_delta;
@ -245,11 +248,13 @@ void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx, float time_delta) {
case State::SENDING: { case State::SENDING: {
tf.ssb.for_each(time_delta, [&](uint16_t id, const std::vector<uint8_t>& data, float& time_since_activity) { tf.ssb.for_each(time_delta, [&](uint16_t id, const std::vector<uint8_t>& data, float& time_since_activity) {
// no ack after 5 sec -> resend // no ack after 5 sec -> resend
if (time_since_activity >= ngc_ft1_ctx->options.sending_resend_without_ack_after) { //if (time_since_activity >= ngc_ft1_ctx->options.sending_resend_without_ack_after) {
if (timeouts_set.count({idx, id})) {
// TODO: can fail // TODO: can fail
_send_pkg_FT1_DATA(tox, group_number, peer_number, idx, id, data.data(), data.size()); _send_pkg_FT1_DATA(tox, group_number, peer_number, idx, id, data.data(), data.size());
peer.cca.onLoss({idx, id}, false); peer.cca.onLoss({idx, id}, false);
time_since_activity = 0.f; time_since_activity = 0.f;
timeouts_set.erase({idx, id});
} }
}); });
@ -261,6 +266,7 @@ void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx, float time_delta) {
// clean up cca // clean up cca
tf.ssb.for_each(time_delta, [&](uint16_t id, const std::vector<uint8_t>& data, float& time_since_activity) { tf.ssb.for_each(time_delta, [&](uint16_t id, const std::vector<uint8_t>& data, float& time_since_activity) {
peer.cca.onLoss({idx, id}, true); peer.cca.onLoss({idx, id}, true);
timeouts_set.erase({idx, id});
}); });
tf_opt.reset(); tf_opt.reset();
@ -271,7 +277,7 @@ void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx, float time_delta) {
// if chunks in flight < window size (2) // if chunks in flight < window size (2)
//while (tf.ssb.size() < ngc_ft1_ctx->options.packet_window_size) { //while (tf.ssb.size() < ngc_ft1_ctx->options.packet_window_size) {
int64_t can_packet_size {peer.cca.canSend()}; int64_t can_packet_size {static_cast<int64_t>(peer.cca.canSend())};
//if (can_packet_size) { //if (can_packet_size) {
//std::cerr << "FT: can_packet_size: " << can_packet_size; //std::cerr << "FT: can_packet_size: " << can_packet_size;
//} //}
@ -283,8 +289,10 @@ void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx, float time_delta) {
//size_t chunk_size = std::min<size_t>(496u, tf.file_size - tf.file_size_current); //size_t chunk_size = std::min<size_t>(496u, tf.file_size - tf.file_size_current);
//size_t chunk_size = std::min<size_t>(can_packet_size, tf.file_size - tf.file_size_current); //size_t chunk_size = std::min<size_t>(can_packet_size, tf.file_size - tf.file_size_current);
size_t chunk_size = std::min<size_t>({ size_t chunk_size = std::min<size_t>({
496u, //496u,
can_packet_size, //996u,
peer.cca.MAXIMUM_SEGMENT_DATA_SIZE,
static_cast<size_t>(can_packet_size),
tf.file_size - tf.file_size_current tf.file_size - tf.file_size_current
}); });
if (chunk_size == 0) { if (chunk_size == 0) {
@ -322,10 +330,12 @@ void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx, float time_delta) {
case State::FINISHING: // we still have unacked packets case State::FINISHING: // we still have unacked packets
tf.ssb.for_each(time_delta, [&](uint16_t id, const std::vector<uint8_t>& data, float& time_since_activity) { tf.ssb.for_each(time_delta, [&](uint16_t id, const std::vector<uint8_t>& data, float& time_since_activity) {
// no ack after 5 sec -> resend // no ack after 5 sec -> resend
if (time_since_activity >= ngc_ft1_ctx->options.sending_resend_without_ack_after) { //if (time_since_activity >= ngc_ft1_ctx->options.sending_resend_without_ack_after) {
if (timeouts_set.count({idx, id})) {
_send_pkg_FT1_DATA(tox, group_number, peer_number, idx, id, data.data(), data.size()); _send_pkg_FT1_DATA(tox, group_number, peer_number, idx, id, data.data(), data.size());
peer.cca.onLoss({idx, id}, false); peer.cca.onLoss({idx, id}, false);
time_since_activity = 0.f; time_since_activity = 0.f;
timeouts_set.erase({idx, id});
} }
}); });
if (tf.time_since_activity >= ngc_ft1_ctx->options.sending_give_up_after) { if (tf.time_since_activity >= ngc_ft1_ctx->options.sending_give_up_after) {
@ -336,6 +346,7 @@ void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx, float time_delta) {
// clean up cca // clean up cca
tf.ssb.for_each(time_delta, [&](uint16_t id, const std::vector<uint8_t>& data, float& time_since_activity) { tf.ssb.for_each(time_delta, [&](uint16_t id, const std::vector<uint8_t>& data, float& time_since_activity) {
peer.cca.onLoss({idx, id}, true); peer.cca.onLoss({idx, id}, true);
timeouts_set.erase({idx, id});
}); });
tf_opt.reset(); tf_opt.reset();
@ -880,7 +891,7 @@ static void _handle_FT1_DATA_ACK(
// delete if all packets acked // delete if all packets acked
if (transfer.file_size == transfer.file_size_current && transfer.ssb.size() == 0) { if (transfer.file_size == transfer.file_size_current && transfer.ssb.size() == 0) {
fprintf(stderr, "FT: %d done\n", transfer_id); fprintf(stderr, "FT: %d done\n", transfer_id);
peer.send_transfers[transfer_id] = std::nullopt; peer.send_transfers[transfer_id].reset();
} }
} }

View File

@ -22,10 +22,8 @@ struct NGC_FT1_options {
float init_retry_timeout_after; // 10sec float init_retry_timeout_after; // 10sec
float sending_resend_without_ack_after; // 5sec //float sending_resend_without_ack_after; // 5sec
float sending_give_up_after; // 30sec float sending_give_up_after; // 30sec
size_t packet_window_size; // 2
}; };
// uint32_t - same as tox friend ft // uint32_t - same as tox friend ft