Green-Sky/solanaceae_ngc_ft1
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Compare commits

base: Green-Sky:4ee5dd6ca5c85d04aaf15d3394f71c10880ab2b5
Branches Tags
Green-Sky:master
Green-Sky:ignore_timed_out_for_in_flight
Green-Sky:broken_somehow
Green-Sky:cca_rework
..
compare: Green-Sky:1c5f4f24a799d78da0a29f83c789a80256fe677a
Branches Tags
Green-Sky:master
Green-Sky:ignore_timed_out_for_in_flight
Green-Sky:broken_somehow
Green-Sky:cca_rework

No commits in common. "4ee5dd6ca5c85d04aaf15d3394f71c10880ab2b5" and "1c5f4f24a799d78da0a29f83c789a80256fe677a" have entirely different histories.
4ee5dd6ca5 ... 1c5f4f24a7

11 changed files with 28 additions and 418 deletions
Show Stats Expand all files Collapse all files

4
CMakeLists.txt
View File

@ -23,12 +23,8 @@ add_library(solanaceae_ngcft1
./solanaceae/ngc_ft1/ngcft1.cpp
./solanaceae/ngc_ft1/cca.hpp
./solanaceae/ngc_ft1/flow_only.hpp
./solanaceae/ngc_ft1/flow_only.cpp
./solanaceae/ngc_ft1/ledbat.hpp
./solanaceae/ngc_ft1/ledbat.cpp
./solanaceae/ngc_ft1/cubic.hpp
./solanaceae/ngc_ft1/cubic.cpp
./solanaceae/ngc_ft1/rcv_buf.hpp
./solanaceae/ngc_ft1/rcv_buf.cpp

13
solanaceae/ngc_ft1/cca.hpp
View File

@ -2,17 +2,6 @@
#include <vector>
#include <cstdint>
#include <cstddef>
// TODO: refactor, more state tracking in ccai and seperate into flow and congestion algos
inline bool isSkipSeqID(const std::pair<uint8_t, uint16_t>& a, const std::pair<uint8_t, uint16_t>& b) {
// this is not perfect, would need more ft id based history
if (a.first != b.first) {
return false; // we dont know
} else {
return a.second+1 != b.second;
}
}
struct CCAI {
public: // config
@ -49,7 +38,7 @@ struct CCAI {
// TODO: api for how much data we should send
// take time since last sent into account
// respect max_byterate_allowed
virtual size_t canSend(void) = 0;
virtual size_t canSend(void) const = 0;
// get the list of timed out seq_ids
virtual std::vector<SeqIDType> getTimeouts(void) const = 0;

62
solanaceae/ngc_ft1/cubic.cpp
View File

@ -1,62 +0,0 @@
#include "./cubic.hpp"
#include <cmath>
#include <iostream>
float CUBIC::getCWnD(void) const {
const double K = cbrt(
(_window_max * (1. - BETA)) / SCALING_CONSTANT
);
const double time_since_reduction = getTimeNow() - _time_point_reduction;
const double TK = time_since_reduction - K;
const double cwnd =
SCALING_CONSTANT
* TK * TK * TK // TK^3
+ _window_max
;
#if 0
std::cout
<< "K:" << K
<< " ts:" << time_since_reduction
<< " TK:" << TK
<< " cwnd:" << cwnd
<< " rtt:" << getCurrentDelay()
<< "\n"
;
#endif
return std::max<float>(cwnd, 2.f * MAXIMUM_SEGMENT_SIZE);
}
void CUBIC::onCongestion(void) {
if (getTimeNow() - _time_point_reduction >= getCurrentDelay()) {
const auto current_cwnd = getCWnD();
_time_point_reduction = getTimeNow();
_window_max = current_cwnd;
std::cout << "CONGESTION! cwnd:" << current_cwnd << "\n";
}
}
size_t CUBIC::canSend(void) {
const auto fspace_pkgs = FlowOnly::canSend();
if (fspace_pkgs == 0u) {
return 0u;
}
const int64_t cspace_bytes = getCWnD() - _in_flight_bytes;
if (cspace_bytes < MAXIMUM_SEGMENT_DATA_SIZE) {
return 0u;
}
// limit to whole packets
size_t cspace_pkgs = std::floor(cspace_bytes / MAXIMUM_SEGMENT_DATA_SIZE) * MAXIMUM_SEGMENT_DATA_SIZE;
return std::min(cspace_pkgs, fspace_pkgs);
}

52
solanaceae/ngc_ft1/cubic.hpp
View File

@ -1,52 +0,0 @@
#pragma once
#include "./flow_only.hpp"
#include <chrono>
struct CUBIC : public FlowOnly {
//using clock = std::chrono::steady_clock;
public: // config
static constexpr float BETA {0.7f};
static constexpr float SCALING_CONSTANT {0.4f};
static constexpr float RTT_EMA_ALPHA = 0.1f; // 0.1 is very smooth, might need more
private:
// window size before last reduciton
double _window_max {2.f * MAXIMUM_SEGMENT_SIZE}; // start with mss*2
//double _window_last_max {2.f * MAXIMUM_SEGMENT_SIZE};
double _time_point_reduction {getTimeNow()};
private:
float getCWnD(void) const;
// moving avg over the last few delay samples
// VERY sensitive to bundling acks
//float getCurrentDelay(void) const;
//void addRTT(float new_delay);
void onCongestion(void) override;
public: // api
CUBIC(size_t maximum_segment_data_size) : FlowOnly(maximum_segment_data_size) {}
// TODO: api for how much data we should send
// take time since last sent into account
// respect max_byterate_allowed
size_t canSend(void) override;
// get the list of timed out seq_ids
//std::vector<SeqIDType> getTimeouts(void) const override;
public: // callbacks
// data size is without overhead
//void onSent(SeqIDType seq, size_t data_size) override;
//void onAck(std::vector<SeqIDType> seqs) override;
// if discard, not resent, not inflight
//void onLoss(SeqIDType seq, bool discard) override;
};

160
solanaceae/ngc_ft1/flow_only.cpp
View File

@ -1,160 +0,0 @@
#include "./flow_only.hpp"
#include <cmath>
#include <cassert>
#include <iostream>
#include <algorithm>
float FlowOnly::getCurrentDelay(void) const {
return std::min(_rtt_ema, RTT_MAX);
}
void FlowOnly::addRTT(float new_delay) {
// lerp(new_delay, rtt_ema, 0.1)
_rtt_ema = RTT_EMA_ALPHA * new_delay + (1.f - RTT_EMA_ALPHA) * _rtt_ema;
}
void FlowOnly::updateWindow(void) {
const float current_delay {getCurrentDelay()};
_fwnd = max_byterate_allowed * current_delay;
//_fwnd *= 1.3f; // try do balance conservative algo a bit, current_delay
_fwnd = std::max(_fwnd, 2.f * MAXIMUM_SEGMENT_DATA_SIZE);
}
size_t FlowOnly::canSend(void) {
if (_in_flight.empty()) {
assert(_in_flight_bytes == 0);
return MAXIMUM_SEGMENT_DATA_SIZE;
}
updateWindow();
const int64_t fspace = _fwnd - _in_flight_bytes;
if (fspace < MAXIMUM_SEGMENT_DATA_SIZE) {
return 0u;
}
// limit to whole packets
size_t space = std::floor(fspace / MAXIMUM_SEGMENT_DATA_SIZE)
* MAXIMUM_SEGMENT_DATA_SIZE;
return space;
}
std::vector<FlowOnly::SeqIDType> FlowOnly::getTimeouts(void) const {
std::vector<SeqIDType> list;
// after 3 rtt delay, we trigger timeout
const auto now_adjusted = getTimeNow() - getCurrentDelay()*3.f;
for (const auto& [seq, time_stamp, size] : _in_flight) {
if (now_adjusted > time_stamp) {
list.push_back(seq);
}
}
return list;
}
void FlowOnly::onSent(SeqIDType seq, size_t data_size) {
if constexpr (true) {
for (const auto& it : _in_flight) {
assert(std::get<0>(it) != seq);
}
}
_in_flight.push_back({seq, getTimeNow(), data_size + SEGMENT_OVERHEAD});
_in_flight_bytes += data_size + SEGMENT_OVERHEAD;
//_recently_sent_bytes += data_size + SEGMENT_OVERHEAD;
}
void FlowOnly::onAck(std::vector<SeqIDType> seqs) {
if (seqs.empty()) {
assert(false && "got empty list of acks???");
return;
}
const auto now {getTimeNow()};
// first seq in seqs is the actual value, all extra are for redundency
{ // skip in ack is congestion event
// 1. look at primary ack of packet
auto it = std::find_if(_in_flight.begin(), _in_flight.end(), [seq = seqs.front()](const auto& v) -> bool {
return std::get<0>(v) == seq;
});
if (it != _in_flight.end()) {
if (it != _in_flight.begin()) {
// not next expected seq -> skip detected
std::cout << "CONGESTION out of order\n";
onCongestion();
//if (getTimeNow() >= _last_congestion_event + _last_congestion_rtt) {
//_recently_lost_data = true;
//_last_congestion_event = getTimeNow();
//_last_congestion_rtt = getCurrentDelay();
//}
} else {
// only mesure delay, if not a congestion
addRTT(now - std::get<1>(*it));
}
} else {
// !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
#if 0
// assume we got a duplicated packet
std::cout << "CONGESTION duplicate\n";
onCongestion();
#endif
}
}
for (const auto& seq : seqs) {
auto it = std::find_if(_in_flight.begin(), _in_flight.end(), [seq](const auto& v) -> bool {
return std::get<0>(v) == seq;
});
if (it == _in_flight.end()) {
continue; // not found, ignore
} else {
//most_recent = std::max(most_recent, std::get<1>(*it));
_in_flight_bytes -= std::get<2>(*it);
assert(_in_flight_bytes >= 0);
//_recently_acked_data += std::get<2>(*it);
_in_flight.erase(it);
}
}
}
void FlowOnly::onLoss(SeqIDType seq, bool discard) {
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;
});
if (it == _in_flight.end()) {
// error
return; // not found, ignore ??
}
std::cerr << "FLOW loss\n";
// "if data lost is not to be retransmitted"
if (discard) {
_in_flight_bytes -= std::get<2>(*it);
assert(_in_flight_bytes >= 0);
_in_flight.erase(it);
}
// TODO: reset timestamp?
#if 0 // temporarily disable ce for timeout
// at most once per rtt?
// TODO: use delay at event instead
if (getTimeNow() >= _last_congestion_event + _last_congestion_rtt) {
_recently_lost_data = true;
_last_congestion_event = getTimeNow();
_last_congestion_rtt = getCurrentDelay();
}
#endif
}

75
solanaceae/ngc_ft1/flow_only.hpp
View File

@ -1,75 +0,0 @@
#pragma once
#include "./cca.hpp"
#include <chrono>
#include <vector>
#include <tuple>
struct FlowOnly : public CCAI {
protected:
using clock = std::chrono::steady_clock;
public: // config
static constexpr float RTT_EMA_ALPHA = 0.1f; // might need over time
static constexpr float RTT_MAX = 2.f; // 2 sec is probably too much
//float max_byterate_allowed {100.f*1024*1024}; // 100MiB/s
float max_byterate_allowed {10.f*1024*1024}; // 10MiB/s
//float max_byterate_allowed {1.f*1024*1024}; // 1MiB/s
//float max_byterate_allowed {0.6f*1024*1024}; // 600KiB/s
//float max_byterate_allowed {0.5f*1024*1024}; // 500KiB/s
//float max_byterate_allowed {0.05f*1024*1024}; // 50KiB/s
//float max_byterate_allowed {0.15f*1024*1024}; // 150KiB/s
protected:
// initialize to low value, will get corrected very fast
float _fwnd {0.01f * max_byterate_allowed}; // in bytes
// rtt exponental moving average
float _rtt_ema {0.1f};
// list of sequence ids and timestamps of when they where sent (and payload size)
std::vector<std::tuple<SeqIDType, float, size_t>> _in_flight;
int64_t _in_flight_bytes {0};
clock::time_point _time_start_offset;
protected:
// make values relative to algo start for readability (and precision)
// get timestamp in seconds
double getTimeNow(void) const {
return std::chrono::duration<double>{clock::now() - _time_start_offset}.count();
}
// moving avg over the last few delay samples
// VERY sensitive to bundling acks
float getCurrentDelay(void) const;
void addRTT(float new_delay);
void updateWindow(void);
virtual void onCongestion(void) {};
public: // api
FlowOnly(size_t maximum_segment_data_size) : CCAI(maximum_segment_data_size) {}
// TODO: api for how much data we should send
// take time since last sent into account
// respect max_byterate_allowed
size_t canSend(void) override;
// get the list of timed out seq_ids
std::vector<SeqIDType> getTimeouts(void) const override;
public: // callbacks
// data size is without overhead
void onSent(SeqIDType seq, size_t data_size) override;
void onAck(std::vector<SeqIDType> seqs) override;
// if discard, not resent, not inflight
void onLoss(SeqIDType seq, bool discard) override;
};

34
solanaceae/ngc_ft1/ledbat.cpp
View File

@ -19,7 +19,7 @@ LEDBAT::LEDBAT(size_t maximum_segment_data_size) : CCAI(maximum_segment_data_siz
_time_start_offset = clock::now();
}
size_t LEDBAT::canSend(void) {
size_t LEDBAT::canSend(void) const {
if (_in_flight.empty()) {
return MAXIMUM_SEGMENT_DATA_SIZE;
}
@ -68,11 +68,6 @@ void LEDBAT::onSent(SeqIDType seq, size_t data_size) {
}
void LEDBAT::onAck(std::vector<SeqIDType> seqs) {
if (seqs.empty()) {
assert(false && "got empty list of acks???");
return;
}
// only take the smallest value
float most_recent {-std::numeric_limits<float>::infinity()};
@ -80,23 +75,6 @@ void LEDBAT::onAck(std::vector<SeqIDType> seqs) {
const auto now {getTimeNow()};
{ // skip in ack is congestion event
// 1. look at primary ack of packet
auto it = std::find_if(_in_flight.begin(), _in_flight.end(), [seq = seqs.front()](const auto& v) -> bool {
return std::get<0>(v) == seq;
});
if (it != _in_flight.end()) {
if (isSkipSeqID(_last_ack_got, std::get<0>(*it))) {
if (getTimeNow() >= _last_congestion_event + _last_congestion_rtt) {
_recently_lost_data = true;
_last_congestion_event = getTimeNow();
_last_congestion_rtt = getCurrentDelay();
}
}
// TODO: only if newer, triggers double event otherwise (without a timer)
_last_ack_got = std::get<0>(*it);
}
}
for (const auto& seq : seqs) {
auto it = std::find_if(_in_flight.begin(), _in_flight.end(), [seq](const auto& v) -> bool {
return std::get<0>(v) == seq;
@ -146,17 +124,15 @@ void LEDBAT::onLoss(SeqIDType seq, bool discard) {
}
// TODO: reset timestamp?
#if 0 // temporarily disable ce for timeout
// at most once per rtt?
// TODO: use delay at event instead
if (getTimeNow() >= _last_congestion_event + _last_congestion_rtt) {
_recently_lost_data = true;
_last_congestion_event = getTimeNow();
_last_congestion_rtt = getCurrentDelay();
}
#endif
updateWindows();
updateWindows();
}
}
float LEDBAT::getCurrentDelay(void) const {
@ -229,7 +205,7 @@ void LEDBAT::updateWindows(void) {
if (_recently_lost_data) {
_cwnd = std::clamp(
_cwnd * 0.7f,
_cwnd / 2.f,
//_cwnd / 1.6f,
2.f * MAXIMUM_SEGMENT_SIZE,
_cwnd
@ -237,7 +213,7 @@ void LEDBAT::updateWindows(void) {
} else {
// LEDBAT++ (the Rethinking the LEDBAT Protocol paper)
// "Multiplicative decrease"
const float constant {1.f}; // spec recs 1
const float constant {2.f}; // spec recs 1
if (queuing_delay < target_delay) {
_cwnd = std::min(
_cwnd + gain,

4
solanaceae/ngc_ft1/ledbat.hpp
View File

@ -61,7 +61,7 @@ struct LEDBAT : public CCAI{
// TODO: api for how much data we should send
// take time since last sent into account
// respect max_byterate_allowed
size_t canSend(void) override;
size_t canSend(void) const override;
// get the list of timed out seq_ids
std::vector<SeqIDType> getTimeouts(void) const override;
@ -123,8 +123,6 @@ struct LEDBAT : public CCAI{
int64_t _in_flight_bytes {0};
SeqIDType _last_ack_got {0xff, 0xffff}; // some default
private: // helper
clock::time_point _time_start_offset;
};

31
solanaceae/ngc_ft1/ngcft1.cpp
View File

@ -143,7 +143,7 @@ bool NGCFT1::sendPKG_FT1_MESSAGE(
return _t.toxGroupSendCustomPacket(group_number, true, pkg) == TOX_ERR_GROUP_SEND_CUSTOM_PACKET_OK;
}
void NGCFT1::updateSendTransfer(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer, size_t idx, std::set<CCAI::SeqIDType>& timeouts_set) {
void NGCFT1::updateSendTransfer(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer, size_t idx, std::set<LEDBAT::SeqIDType>& timeouts_set) {
auto& tf_opt = peer.send_transfers.at(idx);
assert(tf_opt.has_value());
auto& tf = tf_opt.value();
@ -309,7 +309,7 @@ void NGCFT1::updateSendTransfer(float time_delta, uint32_t group_number, uint32_
void NGCFT1::iteratePeer(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer) {
auto timeouts = peer.cca->getTimeouts();
std::set<CCAI::SeqIDType> timeouts_set{timeouts.cbegin(), timeouts.cend()};
std::set<LEDBAT::SeqIDType> timeouts_set{timeouts.cbegin(), timeouts.cend()};
for (size_t idx = 0; idx < peer.send_transfers.size(); idx++) {
if (peer.send_transfers.at(idx).has_value()) {
@ -563,8 +563,7 @@ bool NGCFT1::onEvent(const Events::NGCEXT_ft1_data& e) {
}
// send acks
// reverse, last seq is most recent
std::vector<uint16_t> ack_seq_ids(transfer.rsb.ack_seq_ids.crbegin(), transfer.rsb.ack_seq_ids.crend());
std::vector<uint16_t> ack_seq_ids(transfer.rsb.ack_seq_ids.cbegin(), transfer.rsb.ack_seq_ids.cend());
// TODO: check if this caps at max acks
if (!ack_seq_ids.empty()) {
// TODO: check return value
@ -589,7 +588,7 @@ bool NGCFT1::onEvent(const Events::NGCEXT_ft1_data& e) {
bool NGCFT1::onEvent(const Events::NGCEXT_ft1_data_ack& e) {
#if !NDEBUG
//std::cout << "NGCFT1: FT1_DATA_ACK\n";
std::cout << "NGCFT1: FT1_DATA_ACK\n";
#endif
if (!groups.count(e.group_number)) {
@ -611,17 +610,20 @@ bool NGCFT1::onEvent(const Events::NGCEXT_ft1_data_ack& e) {
return true;
}
//if ((length - curser) % sizeof(uint16_t) != 0) {
//fprintf(stderr, "FT: data_ack with misaligned data\n");
//return;
//}
transfer.time_since_activity = 0.f;
{
std::vector<CCAI::SeqIDType> seqs;
for (const auto it : e.sequence_ids) {
// TODO: improve this o.o
seqs.push_back({e.transfer_id, it});
transfer.ssb.erase(it);
}
peer.cca->onAck(std::move(seqs));
std::vector<LEDBAT::SeqIDType> seqs;
for (const auto it : e.sequence_ids) {
// TODO: improve this o.o
seqs.push_back({e.transfer_id, it});
transfer.ssb.erase(it);
}
peer.cca->onAck(seqs);
// delete if all packets acked
if (transfer.file_size == transfer.file_size_current && transfer.ssb.size() == 0) {
@ -633,7 +635,6 @@ bool NGCFT1::onEvent(const Events::NGCEXT_ft1_data_ack& e) {
e.transfer_id,
}
);
// TODO: check for FINISHING state
peer.send_transfers[e.transfer_id].reset();
}
@ -711,7 +712,7 @@ bool NGCFT1::onToxEvent(const Tox_Event_Group_Peer_Exit* e) {
}
// reset cca
peer.cca = std::make_unique<CUBIC>(500-4); // TODO: replace with tox_group_max_custom_lossy_packet_length()-4
peer.cca = std::make_unique<LEDBAT>(500-4); // TODO: replace with tox_group_max_custom_lossy_packet_length()-4
return false;
}

8
solanaceae/ngc_ft1/ngcft1.hpp
View File

@ -6,9 +6,7 @@
#include <solanaceae/toxcore/tox_event_interface.hpp>
#include <solanaceae/ngc_ext/ngcext.hpp>
#include "./cubic.hpp"
//#include "./flow_only.hpp"
//#include "./ledbat.hpp"
#include "./ledbat.hpp"
#include "./rcv_buf.hpp"
#include "./snd_buf.hpp"
@ -141,7 +139,7 @@ class NGCFT1 : public ToxEventI, public NGCEXTEventI, public NGCFT1EventProvider
struct Group {
struct Peer {
std::unique_ptr<CCAI> cca = std::make_unique<CUBIC>(500-4); // TODO: replace with tox_group_max_custom_lossy_packet_length()-4
std::unique_ptr<CCAI> cca = std::make_unique<LEDBAT>(500-4); // TODO: replace with tox_group_max_custom_lossy_packet_length()-4
struct RecvTransfer {
uint32_t file_kind;
@ -201,7 +199,7 @@ class NGCFT1 : public ToxEventI, public NGCEXTEventI, public NGCFT1EventProvider
bool sendPKG_FT1_DATA_ACK(uint32_t group_number, uint32_t peer_number, uint8_t transfer_id, const uint16_t* seq_ids, size_t seq_ids_size);
bool sendPKG_FT1_MESSAGE(uint32_t group_number, uint32_t message_id, uint32_t file_kind, const uint8_t* file_id, size_t file_id_size);
void updateSendTransfer(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer, size_t idx, std::set<CCAI::SeqIDType>& timeouts_set);
void updateSendTransfer(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer, size_t idx, std::set<LEDBAT::SeqIDType>& timeouts_set);
void iteratePeer(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer);
public:

3
solanaceae/ngc_ft1_sha1/sha1_ngcft1.cpp
View File

@ -826,7 +826,6 @@ bool SHA1_NGCFT1::onEvent(const Events::NGCFT1_send_data& e) {
}
auto& transfer = peer.at(e.transfer_id);
transfer.time_since_activity = 0.f;
if (std::holds_alternative<SendingTransfer::Info>(transfer.v)) {
auto& info_transfer = std::get<SendingTransfer::Info>(transfer.v);
for (size_t i = 0; i < e.data_size && (i + e.data_offset) < info_transfer.info_data.size(); i++) {
@ -860,6 +859,8 @@ bool SHA1_NGCFT1::onEvent(const Events::NGCFT1_send_data& e) {
assert(false && "not implemented?");
}
transfer.time_since_activity = 0.f;
return true;
}