fix missing include for gcc12

ledbat.cpp

@@ -0,0 +1,251 @@
+#include "./ledbat.hpp"
+
+#include <algorithm>
+#include <chrono>
+#include <cmath>
+#include <deque>
+#include <cstdint>
+#include <tuple>
+#include <cassert>
+
+#include <iomanip>
+#include <iostream>
+#include <limits>
+
+// https://youtu.be/0HRwNSA-JYM
+
+inline constexpr bool PLOTTING = false;
+
+LEDBAT::LEDBAT(size_t maximum_segment_data_size) : MAXIMUM_SEGMENT_DATA_SIZE(maximum_segment_data_size) {
+	_time_start_offset = clock::now();
+}
+
+size_t LEDBAT::canSend(void) const {
+	if (_in_flight.empty()) {
+		return MAXIMUM_SEGMENT_DATA_SIZE;
+	}
+
+	const int64_t cspace = _cwnd - _in_flight_bytes;
+	if (cspace < MAXIMUM_SEGMENT_DATA_SIZE) {
+		return 0u;
+	}
+
+	const int64_t fspace = _fwnd - _in_flight_bytes;
+	if (fspace < MAXIMUM_SEGMENT_DATA_SIZE) {
+		return 0u;
+	}
+
+	size_t space = std::ceil(std::min<float>(cspace, fspace) / MAXIMUM_SEGMENT_DATA_SIZE) * MAXIMUM_SEGMENT_DATA_SIZE;
+
+	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) {
+	if (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 LEDBAT::onAck(std::vector<SeqIDType> seqs) {
+	// only take the smallest value
+	float most_recent {-std::numeric_limits<float>::infinity()};
+
+	int64_t acked_data {0};
+
+	const auto now {getTimeNow()};
+
+	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 {
+			addRTT(now - std::get<1>(*it));
+
+			// TODO: remove
+			most_recent = std::max(most_recent, std::get<1>(*it));
+			_in_flight_bytes -= std::get<2>(*it);
+			_recently_acked_data += std::get<2>(*it);
+			assert(_in_flight_bytes >= 0); // TODO: this triggers
+			_in_flight.erase(it);
+		}
+	}
+
+	if (most_recent == -std::numeric_limits<float>::infinity()) {
+		return; // not found, ignore
+	}
+
+	updateWindows();
+}
+
+void LEDBAT::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 ??
+	}
+
+	_recently_lost_data = true;
+
+	// at most once per rtt?
+
+	if (PLOTTING) {
+		std::cerr << "CCA: onLoss: TIME: " << getTimeNow() << "\n";
+	}
+
+	// TODO: "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?
+
+	updateWindows();
+}
+
+float LEDBAT::getCurrentDelay(void) const {
+	float sum {0.f};
+	size_t count {0};
+	for (size_t i = 0; i < _tmp_rtt_buffer.size(); i++) {
+		//sum += _tmp_rtt_buffer.at(_tmp_rtt_buffer.size()-(1+i));
+		sum += _tmp_rtt_buffer.at(i);
+		count++;
+	}
+
+	if (count) {
+		return sum / count;
+	} else {
+		return std::numeric_limits<float>::infinity();
+	}
+}
+
+void LEDBAT::addRTT(float new_delay) {
+	auto now = getTimeNow();
+
+	_base_delay = std::min(_base_delay, new_delay);
+	// TODO: use fixed size instead? allocations can ruin perf
+	_rtt_buffer.push_back({now, new_delay});
+
+	_tmp_rtt_buffer.push_front(new_delay);
+	// HACKY
+	if (_tmp_rtt_buffer.size() > current_delay_filter_window) {
+		_tmp_rtt_buffer.resize(current_delay_filter_window);
+	}
+
+	// is it 1 minute yet
+	if (now - _rtt_buffer.front().first >= 30.f) {
+
+		float new_section_minimum = new_delay;
+		for (const auto it : _rtt_buffer) {
+			new_section_minimum = std::min(it.second, new_section_minimum);
+		}
+
+		_rtt_buffer_minutes.push_back(new_section_minimum);
+
+		_rtt_buffer.clear();
+
+		if (_rtt_buffer_minutes.size() > 20) {
+			_rtt_buffer_minutes.pop_front();
+		}
+
+		_base_delay = std::numeric_limits<float>::infinity();
+		for (const float it : _rtt_buffer_minutes) {
+			_base_delay = std::min(_base_delay, it);
+		}
+	}
+}
+
+void LEDBAT::updateWindows(void) {
+	const auto now {getTimeNow()};
+
+	const float current_delay {getCurrentDelay()};
+
+	if (now - _last_cwnd >= current_delay) {
+		const float queuing_delay {current_delay - _base_delay};
+
+		_fwnd = max_byterate_allowed * current_delay;
+		_fwnd *= 1.3f; // try do balance conservative algo a bit, current_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 *= _recently_acked_data/5.f; // from packets to bytes ~
+		//gain *= 0.1f;
+
+		if (_recently_lost_data) {
+			_cwnd = std::clamp(
+				_cwnd / 2.f,
+				//_cwnd / 1.6f,
+				2.f * MAXIMUM_SEGMENT_SIZE,
+				_cwnd
+			);
+		} else {
+			// LEDBAT++ (the Rethinking the LEDBAT Protocol paper)
+			// "Multiplicative decrease"
+			const float constant {2.f}; // spec recs 1
+			if (queuing_delay < target_delay) {
+				_cwnd = std::min(
+					_cwnd + gain,
+					_fwnd
+				);
+			} else if (queuing_delay > target_delay) {
+				_cwnd = std::clamp(
+					_cwnd + std::max(
+						gain - constant * _cwnd * (queuing_delay / target_delay - 1.f),
+						-_cwnd/2.f // at most halve
+					),
+
+					// never drop below 2 "packets" in flight
+					2.f * MAXIMUM_SEGMENT_SIZE,
+
+					// cap rate
+					_fwnd
+				);
+			} // no else, we on point. very unlikely with float
+		}
+
+		if (PLOTTING) { // plotting
+			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 << " current_delay: " << current_delay << "\n";
+			std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " base_delay: " << _base_delay << "\n";
+			std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " gain: " << gain << "\n";
+			std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " speed: " << (_recently_sent_bytes / (now - _last_cwnd)) / (1024*1024) << "\n";
+			std::cerr << std::fixed << "CCA: onAck: TIME: " << now << " in_flight_bytes: " << _in_flight_bytes << "\n";
+		}
+
+		_last_cwnd = now;
+		_recently_acked_data = 0;
+		_recently_lost_data = false;
+		_recently_sent_bytes = 0;
+	}
+}
+

ledbat.hpp

@@ -0,0 +1,122 @@
+#pragma once
+
+#include <chrono>
+#include <deque>
+#include <vector>
+#include <cstdint>
+
+// LEDBAT: https://www.rfc-editor.org/rfc/rfc6817
+// LEDBAT++: https://www.ietf.org/archive/id/draft-irtf-iccrg-ledbat-plus-plus-01.txt
+
+// LEDBAT++ implementation
+struct LEDBAT {
+	public: // config
+		using SeqIDType = std::pair<uint8_t, uint16_t>; // tf_id, seq_id
+
+		static constexpr size_t IPV4_HEADER_SIZE {20};
+		static constexpr size_t IPV6_HEADER_SIZE {40}; // bru
+		static constexpr size_t UDP_HEADER_SIZE {8};
+
+		// TODO: tcp AND IPv6 will be different
+		static constexpr size_t SEGMENT_OVERHEAD {
+			4+ // ft overhead
+			46+ // tox?
+			UDP_HEADER_SIZE+
+			IPV4_HEADER_SIZE
+		};
+
+		// TODO: make configurable, set with tox ngc lossy packet size
+		//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
+		//const float target_delay {0.060f};
+		const float target_delay {0.030f};
+		//const float target_delay {0.120f}; // 2x if relayed?
+
+		// TODO: use a factor for multiple of rtt
+		static constexpr size_t current_delay_filter_window {16*4};
+
+		//static constexpr size_t rtt_buffer_size_max {2000};
+
+		float max_byterate_allowed {10*1024*1024}; // 10MiB/s
+
+	public:
+		LEDBAT(size_t maximum_segment_data_size);
+
+		// return the current believed window in bytes of how much data can be inflight,
+		// without overstepping the delay requirement
+		float getCWnD(void) const {
+			return _cwnd;
+		}
+
+		// TODO: api for how much data we should send
+		// take time since last sent into account
+		// respect max_byterate_allowed
+		size_t canSend(void) const;
+
+		// get the list of timed out seq_ids
+		std::vector<SeqIDType> getTimeouts(void) const;
+
+	public: // callbacks
+		// data size is without overhead
+		void onSent(SeqIDType seq, size_t data_size);
+
+		void onAck(std::vector<SeqIDType> seqs);
+
+		// if discard, not resent, not inflight
+		void onLoss(SeqIDType seq, bool discard);
+
+	private:
+		using clock = std::chrono::steady_clock;
+
+		// make values relative to algo start for readability (and precision)
+		// get timestamp in seconds
+		float getTimeNow(void) const {
+			return std::chrono::duration<float>{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 updateWindows(void);
+
+	private: // state
+		//float _cto {2.f}; // congestion timeout value in seconds
+
+		float _cwnd {2.f * MAXIMUM_SEGMENT_SIZE}; // in bytes
+		float _base_delay {2.f}; // lowest mesured delay in _rtt_buffer in seconds
+
+		float _last_cwnd {0.f}; // timepoint of last cwnd correction
+		int64_t _recently_acked_data {0}; // reset on _last_cwnd
+		bool _recently_lost_data {false};
+		int64_t _recently_sent_bytes {0};
+
+		// initialize to low value, will get corrected very fast
+		float _fwnd {0.01f * max_byterate_allowed}; // in bytes
+
+
+		// ssthresh
+
+		// spec recomends 10min
+		// TODO: optimize and devide into spans of 1min (spec recom)
+		std::deque<float> _tmp_rtt_buffer;
+		std::deque<std::pair<float, float>> _rtt_buffer; // timepoint, delay
+		std::deque<float> _rtt_buffer_minutes;
+
+		// list of sequence ids and timestamps of when they where sent
+		std::deque<std::tuple<SeqIDType, float, size_t>> _in_flight;
+
+		int64_t _in_flight_bytes {0};
+
+	private: // helper
+		clock::time_point _time_start_offset;
+};
+

ngc_ft1.cpp

@@ -2,14 +2,20 @@
 
 #include "ngc_ext.hpp"
 
+#include "./ledbat.hpp"
+
+#include <algorithm>
 #include <vector>
 #include <array>
 #include <deque>
 // TODO: should i really use both?
 #include <unordered_map>
 #include <map>
+#include <set>
 #include <optional>
 #include <cassert>
+#include <cstdio>
+#include <iostream>
 
 struct SendSequenceBuffer {
 	struct SSBEntry {
@@ -70,7 +76,7 @@ struct RecvSequenceBuffer {
 	void add(uint16_t seq_id, std::vector<uint8_t>&& data) {
 		entries[seq_id] = {data};
 		ack_seq_ids.push_back(seq_id);
-		if (ack_seq_ids.size() > 5) { // TODO: magic
+		if (ack_seq_ids.size() > 3) { // TODO: magic
 			ack_seq_ids.pop_front();
 		}
 	}
@@ -102,19 +108,21 @@ struct RecvSequenceBuffer {
 struct NGC_FT1 {
 	NGC_FT1_options options;
 
-	std::unordered_map<NGC_FT1_file_kind, NGC_FT1_recv_request_cb*> cb_recv_request;
-	std::unordered_map<NGC_FT1_file_kind, NGC_FT1_recv_init_cb*> cb_recv_init;
-	std::unordered_map<NGC_FT1_file_kind, NGC_FT1_recv_data_cb*> cb_recv_data;
-	std::unordered_map<NGC_FT1_file_kind, NGC_FT1_send_data_cb*> cb_send_data;
-	std::unordered_map<NGC_FT1_file_kind, void*> ud_recv_request;
-	std::unordered_map<NGC_FT1_file_kind, void*> ud_recv_init;
-	std::unordered_map<NGC_FT1_file_kind, void*> ud_recv_data;
-	std::unordered_map<NGC_FT1_file_kind, void*> ud_send_data;
+	std::unordered_map<uint32_t, NGC_FT1_recv_request_cb*> cb_recv_request;
+	std::unordered_map<uint32_t, NGC_FT1_recv_init_cb*> cb_recv_init;
+	std::unordered_map<uint32_t, NGC_FT1_recv_data_cb*> cb_recv_data;
+	std::unordered_map<uint32_t, NGC_FT1_send_data_cb*> cb_send_data;
+	std::unordered_map<uint32_t, void*> ud_recv_request;
+	std::unordered_map<uint32_t, void*> ud_recv_init;
+	std::unordered_map<uint32_t, void*> ud_recv_data;
+	std::unordered_map<uint32_t, void*> ud_send_data;
 
 	struct Group {
 		struct Peer {
+			LEDBAT cca{500-4}; // TODO: replace with tox_group_max_custom_lossy_packet_length()-4
+
 			struct RecvTransfer {
-				NGC_FT1_file_kind file_kind;
+				uint32_t file_kind;
 				std::vector<uint8_t> file_id;
 
 				enum class State {
@@ -133,7 +141,7 @@ struct NGC_FT1 {
 			size_t next_recv_transfer_idx {0}; // next id will be 0
 
 			struct SendTransfer {
-				NGC_FT1_file_kind file_kind;
+				uint32_t file_kind;
 				std::vector<uint8_t> file_id;
 
 				enum class State {
@@ -141,11 +149,8 @@ struct NGC_FT1 {
 
 					SENDING, // we got the ack and are now sending data
 
-					// is this real?
 					FINISHING, // we sent all data but acks still outstanding????
 
-					FINFIN, // we sent the data_fin and are waiting for the data_fin_ack
-
 					// delete
 				} state;
 
@@ -168,8 +173,8 @@ struct NGC_FT1 {
 };
 
 // send pkgs
-static bool _send_pkg_FT1_REQUEST(const Tox* tox, uint32_t group_number, uint32_t peer_number, uint8_t file_kind, const uint8_t* file_id, size_t file_id_size);
-static bool _send_pkg_FT1_INIT(const Tox* tox, uint32_t group_number, uint32_t peer_number, uint8_t file_kind, uint64_t file_size, uint8_t transfer_id, const uint8_t* file_id, size_t file_id_size);
+static bool _send_pkg_FT1_REQUEST(const Tox* tox, uint32_t group_number, uint32_t peer_number, uint32_t file_kind, const uint8_t* file_id, size_t file_id_size);
+static bool _send_pkg_FT1_INIT(const Tox* tox, uint32_t group_number, uint32_t peer_number, uint32_t file_kind, uint64_t file_size, uint8_t transfer_id, const uint8_t* file_id, size_t file_id_size);
 static bool _send_pkg_FT1_INIT_ACK(const Tox* tox, uint32_t group_number, uint32_t peer_number, uint8_t transfer_id);
 static bool _send_pkg_FT1_DATA(const Tox* tox, uint32_t group_number, uint32_t peer_number, uint8_t transfer_id, uint16_t sequence_id, const uint8_t* data, size_t data_size);
 static bool _send_pkg_FT1_DATA_ACK(const Tox* tox, uint32_t group_number, uint32_t peer_number, uint8_t transfer_id, const uint16_t* seq_ids, size_t seq_ids_size);
@@ -207,24 +212,25 @@ void NGC_FT1_kill(NGC_FT1* ngc_ft1_ctx) {
 	delete ngc_ft1_ctx;
 }
 
-void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx) {
-//void NGC_FT1_iterate(Tox *tox, NGC_EXT_CTX* ngc_ext_ctx/*, void *user_data*/) {
+void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx, float time_delta) {
 	assert(ngc_ft1_ctx);
 
 	for (auto& [group_number, group] : ngc_ft1_ctx->groups) {
 		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++) {
 				auto& tf_opt = peer.send_transfers[idx];
-				if (tf_opt) {
+				if (tf_opt.has_value()) {
 					auto& tf = tf_opt.value();
 
-					tf.time_since_activity += 0.025f; // TODO: actual delta
+					tf.time_since_activity += time_delta;
 
 					switch (tf.state) {
 						using State = NGC_FT1::Group::Peer::SendTransfer::State;
 						case State::INIT_SENT:
-							if (tf.time_since_activity >= 20.f) {
+							if (tf.time_since_activity >= ngc_ft1_ctx->options.init_retry_timeout_after) {
 								if (tf.inits_sent >= 3) {
 									// delete, timed out 3 times
 									fprintf(stderr, "FT: warning, ft init timed out, deleting\n");
@@ -240,31 +246,57 @@ void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx) {
 							}
 							break;
 						case State::SENDING: {
-								tf.ssb.for_each(0.025f, [&](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
-									if (time_since_activity >= 5.f) {
+									//if (time_since_activity >= ngc_ft1_ctx->options.sending_resend_without_ack_after) {
+									if (timeouts_set.count({idx, id})) {
+										// TODO: can fail
 										_send_pkg_FT1_DATA(tox, group_number, peer_number, idx, id, data.data(), data.size());
+										peer.cca.onLoss({idx, id}, false);
 										time_since_activity = 0.f;
+										timeouts_set.erase({idx, id});
 									}
 								});
 
-								if (tf.time_since_activity >= 30.f) {
+								if (tf.time_since_activity >= ngc_ft1_ctx->options.sending_give_up_after) {
 									// no ack after 30sec, close ft
 									// TODO: notify app
 									fprintf(stderr, "FT: warning, sending ft in progress timed out, deleting\n");
+
+									// clean up cca
+									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);
+										timeouts_set.erase({idx, id});
+									});
+
 									tf_opt.reset();
 									continue; // dangerous control flow
 								}
 
 								assert(ngc_ft1_ctx->cb_send_data.count(tf.file_kind));
 
-								// if chunks in flight < window size (1 lol)
-								while (tf.ssb.size() < 1) {
+								// if chunks in flight < window size (2)
+								//while (tf.ssb.size() < ngc_ft1_ctx->options.packet_window_size) {
+								int64_t can_packet_size {static_cast<int64_t>(peer.cca.canSend())};
+								//if (can_packet_size) {
+									//std::cerr << "FT: can_packet_size: " << can_packet_size;
+								//}
+								size_t count {0};
+								while (can_packet_size > 0 && tf.file_size > 0) {
 									std::vector<uint8_t> new_data;
 
-									size_t chunk_size = std::min<size_t>(400u, tf.file_size - tf.file_size_current);
+									// TODO: parameterize packet size? -> only if JF increases lossy packet size >:)
+									//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>({
+										//496u,
+										//996u,
+										peer.cca.MAXIMUM_SEGMENT_DATA_SIZE,
+										static_cast<size_t>(can_packet_size),
+										tf.file_size - tf.file_size_current
+									});
 									if (chunk_size == 0) {
-										// TODO: set to finishing?
+										tf.state = State::FINISHING;
 										break; // we done
 									}
 
@@ -280,18 +312,46 @@ void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx) {
 									);
 									uint16_t seq_id = tf.ssb.add(std::move(new_data));
 									_send_pkg_FT1_DATA(tox, group_number, peer_number, idx, seq_id, tf.ssb.entries.at(seq_id).data.data(), tf.ssb.entries.at(seq_id).data.size());
+									peer.cca.onSent({idx, seq_id}, chunk_size);
 
+#if defined(EXTRA_LOGGING) && EXTRA_LOGGING == 1
 									fprintf(stderr, "FT: sent data size: %ld (seq %d)\n", chunk_size, seq_id);
+#endif
 
 									tf.file_size_current += chunk_size;
+									can_packet_size -= chunk_size;
+									count++;
 								}
+								//if (count) {
+									//std::cerr << " split over " << count << "\n";
+								//}
 							}
 							break;
-						case State::FINISHING:
+						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) {
+								// no ack after 5 sec -> resend
+								//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());
+									peer.cca.onLoss({idx, id}, false);
+									time_since_activity = 0.f;
+									timeouts_set.erase({idx, id});
+								}
+							});
+							if (tf.time_since_activity >= ngc_ft1_ctx->options.sending_give_up_after) {
+								// no ack after 30sec, close ft
+								// TODO: notify app
+								fprintf(stderr, "FT: warning, sending ft finishing timed out, deleting\n");
+
+								// clean up cca
+								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);
+									timeouts_set.erase({idx, id});
+								});
+
+								tf_opt.reset();
+							}
 							break;
-
-						// finfin o.o
-
 						default: // invalid state, delete
 							fprintf(stderr, "FT: error, ft in invalid state, deleting\n");
 							tf_opt.reset();
@@ -305,7 +365,7 @@ void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx) {
 
 void NGC_FT1_register_callback_recv_request(
 	NGC_FT1* ngc_ft1_ctx,
-	NGC_FT1_file_kind file_kind,
+	uint32_t file_kind,
 	NGC_FT1_recv_request_cb* callback,
 	void* user_data
 ) {
@@ -317,7 +377,7 @@ void NGC_FT1_register_callback_recv_request(
 
 void NGC_FT1_register_callback_recv_init(
 	NGC_FT1* ngc_ft1_ctx,
-	NGC_FT1_file_kind file_kind,
+	uint32_t file_kind,
 	NGC_FT1_recv_init_cb* callback,
 	void* user_data
 ) {
@@ -329,7 +389,7 @@ void NGC_FT1_register_callback_recv_init(
 
 void NGC_FT1_register_callback_recv_data(
 	NGC_FT1* ngc_ft1_ctx,
-	NGC_FT1_file_kind file_kind,
+	uint32_t file_kind,
 	NGC_FT1_recv_data_cb* callback,
 	void* user_data
 ) {
@@ -341,7 +401,7 @@ void NGC_FT1_register_callback_recv_data(
 
 void NGC_FT1_register_callback_send_data(
 	NGC_FT1* ngc_ft1_ctx,
-	NGC_FT1_file_kind file_kind,
+	uint32_t file_kind,
 	NGC_FT1_send_data_cb* callback,
 	void* user_data
 ) {
@@ -357,7 +417,7 @@ void NGC_FT1_send_request_private(
 	uint32_t group_number,
 	uint32_t peer_number,
 
-	NGC_FT1_file_kind file_kind,
+	uint32_t file_kind,
 
 	const uint8_t* file_id,
 	size_t file_id_size
@@ -373,13 +433,13 @@ void NGC_FT1_send_request_private(
 bool NGC_FT1_send_init_private(
 	Tox *tox, NGC_FT1* ngc_ft1_ctx,
 	uint32_t group_number, uint32_t peer_number,
-	NGC_FT1_file_kind file_kind,
+	uint32_t file_kind,
 	const uint8_t* file_id, size_t file_id_size,
 	size_t file_size,
 	uint8_t* transfer_id
 ) {
 	//fprintf(stderr, "TODO: init ft for %08X\n", msg_id);
-	fprintf(stderr, "FT: init ft\n");
+	//fprintf(stderr, "FT: init ft\n");
 
 	if (tox_group_peer_get_connection_status(tox, group_number, peer_number, nullptr) == TOX_CONNECTION_NONE) {
 		fprintf(stderr, "FT: error: cant init ft, peer offline\n");
@@ -430,13 +490,15 @@ bool NGC_FT1_send_init_private(
 	return true;
 }
 
-static bool _send_pkg_FT1_REQUEST(const Tox* tox, uint32_t group_number, uint32_t peer_number, uint8_t file_kind, const uint8_t* file_id, size_t file_id_size) {
+static bool _send_pkg_FT1_REQUEST(const Tox* tox, uint32_t group_number, uint32_t peer_number, uint32_t file_kind, const uint8_t* file_id, size_t file_id_size) {
 	// - 1 byte packet id
-	// - 1 byte (TODO: more?) file_kind
+	// - 4 byte file_kind
 	// - X bytes file_id
 	std::vector<uint8_t> pkg;
 	pkg.push_back(NGC_EXT::FT1_REQUEST);
-	pkg.push_back(file_kind);
+	for (size_t i = 0; i < sizeof(file_kind); i++) {
+		pkg.push_back((file_kind>>(i*8)) & 0xff);
+	}
 	for (size_t i = 0; i < file_id_size; i++) {
 		pkg.push_back(file_id[i]);
 	}
@@ -445,16 +507,18 @@ static bool _send_pkg_FT1_REQUEST(const Tox* tox, uint32_t group_number, uint32_
 	return tox_group_send_custom_private_packet(tox, group_number, peer_number, true, pkg.data(), pkg.size(), nullptr);
 }
 
-static bool _send_pkg_FT1_INIT(const Tox* tox, uint32_t group_number, uint32_t peer_number, uint8_t file_kind, uint64_t file_size, uint8_t transfer_id, const uint8_t* file_id, size_t file_id_size) {
+static bool _send_pkg_FT1_INIT(const Tox* tox, uint32_t group_number, uint32_t peer_number, uint32_t file_kind, uint64_t file_size, uint8_t transfer_id, const uint8_t* file_id, size_t file_id_size) {
 	// - 1 byte packet id
-	// - 1 byte (file_kind)
+	// - 4 byte (file_kind)
 	// - 8 bytes (data size)
 	// - 1 byte (temporary_file_tf_id, for this peer only, technically just a prefix to distinguish between simultainious fts)
 	// - X bytes (file_kind dependent id, differnt sizes)
 
 	std::vector<uint8_t> pkg;
 	pkg.push_back(NGC_EXT::FT1_INIT);
-	pkg.push_back(file_kind);
+	for (size_t i = 0; i < sizeof(file_kind); i++) {
+		pkg.push_back((file_kind>>(i*8)) & 0xff);
+	}
 	for (size_t i = 0; i < sizeof(file_size); i++) {
 		pkg.push_back((file_size>>(i*8)) & 0xff);
 	}
@@ -496,8 +560,8 @@ static bool _send_pkg_FT1_DATA(const Tox* tox, uint32_t group_number, uint32_t p
 		pkg.push_back(data[i]);
 	}
 
-	// lossless?
-	return tox_group_send_custom_private_packet(tox, group_number, peer_number, true, pkg.data(), pkg.size(), nullptr);
+	// lossy
+	return tox_group_send_custom_private_packet(tox, group_number, peer_number, false, pkg.data(), pkg.size(), nullptr);
 }
 
 static bool _send_pkg_FT1_DATA_ACK(const Tox* tox, uint32_t group_number, uint32_t peer_number, uint8_t transfer_id, const uint16_t* seq_ids, size_t seq_ids_size) {
@@ -511,8 +575,8 @@ static bool _send_pkg_FT1_DATA_ACK(const Tox* tox, uint32_t group_number, uint32
 		pkg.push_back((seq_ids[i] >> (1*8)) & 0xff);
 	}
 
-	// lossless?
-	return tox_group_send_custom_private_packet(tox, group_number, peer_number, true, pkg.data(), pkg.size(), nullptr);
+	// lossy
+	return tox_group_send_custom_private_packet(tox, group_number, peer_number, false, pkg.data(), pkg.size(), nullptr);
 }
 
 #define _DATA_HAVE(x, error) if ((length - curser) < (x)) { error; }
@@ -531,13 +595,13 @@ static void _handle_FT1_REQUEST(
 	NGC_FT1* ngc_ft1_ctx = static_cast<NGC_FT1*>(user_data);
 	size_t curser = 0;
 
-	// TODO: might be uint16_t or even larger
-	uint8_t file_kind_u8;
-	_DATA_HAVE(sizeof(file_kind_u8), fprintf(stderr, "FT: packet too small, missing file_kind\n"); return)
-	file_kind_u8 = data[curser++];
-	auto file_kind = static_cast<NGC_FT1_file_kind>(file_kind_u8);
+	uint32_t file_kind {0u};
+	_DATA_HAVE(sizeof(file_kind), fprintf(stderr, "FT: packet too small, missing file_kind\n"); return)
+	for (size_t i = 0; i < sizeof(file_kind); i++, curser++) {
+		file_kind |= uint32_t(data[curser]) << (i*8);
+	}
 
-	fprintf(stderr, "FT: got FT request with file_kind %u [", file_kind_u8);
+	fprintf(stderr, "FT: got FT request with file_kind %u [", file_kind);
 	for (size_t curser_copy = curser; curser_copy < length; curser_copy++) {
 		fprintf(stderr, "%02X", data[curser_copy]);
 	}
@@ -572,13 +636,12 @@ static void _handle_FT1_INIT(
 	NGC_FT1* ngc_ft1_ctx = static_cast<NGC_FT1*>(user_data);
 	size_t curser = 0;
 
-	// - 1 byte (file_kind)
-
-	// TODO: might be uint16_t or even larger
-	uint8_t file_kind_u8;
-	_DATA_HAVE(sizeof(file_kind_u8), fprintf(stderr, "FT: packet too small, missing file_kind\n"); return)
-	file_kind_u8 = data[curser++];
-	auto file_kind = static_cast<NGC_FT1_file_kind>(file_kind_u8);
+	// - 4 byte (file_kind)
+	uint32_t file_kind {0u};
+	_DATA_HAVE(sizeof(file_kind), fprintf(stderr, "FT: packet too small, missing file_kind\n"); return)
+	for (size_t i = 0; i < sizeof(file_kind); i++, curser++) {
+		file_kind |= uint32_t(data[curser]) << (i*8);
+	}
 
 	// - 8 bytes (data size)
 	size_t file_size {0u};
@@ -595,7 +658,7 @@ static void _handle_FT1_INIT(
 	// - X bytes (file_kind dependent id, differnt sizes)
 
 	const std::vector file_id(data+curser, data+curser+(length-curser));
-	fprintf(stderr, "FT: got FT init with file_kind:%u file_size:%lu tf_id:%u [", file_kind_u8, file_size, transfer_id);
+	fprintf(stderr, "FT: got FT init with file_kind:%u file_size:%lu tf_id:%u [", file_kind, file_size, transfer_id);
 	for (size_t curser_copy = curser; curser_copy < length; curser_copy++) {
 		fprintf(stderr, "%02X", data[curser_copy]);
 	}
@@ -623,7 +686,9 @@ static void _handle_FT1_INIT(
 
 	if (accept_ft) {
 		_send_pkg_FT1_INIT_ACK(tox, group_number, peer_number, transfer_id);
+#if defined(EXTRA_LOGGING) && EXTRA_LOGGING == 1
 		fprintf(stderr, "FT: accepted init\n");
+#endif
 		auto& peer = ngc_ft1_ctx->groups[group_number].peers[peer_number];
 		if (peer.recv_transfers[transfer_id].has_value()) {
 			fprintf(stderr, "FT: overwriting existing recv_transfer %d\n", transfer_id);
@@ -799,8 +864,8 @@ static void _handle_FT1_DATA_ACK(
 	NGC_FT1::Group::Peer::SendTransfer& transfer = peer.send_transfers[transfer_id].value();
 
 	using State = NGC_FT1::Group::Peer::SendTransfer::State;
-	if (transfer.state != State::SENDING) {
-		fprintf(stderr, "FT: data_ack but not in SENDING state\n");
+	if (transfer.state != State::SENDING && transfer.state != State::FINISHING) {
+		fprintf(stderr, "FT: data_ack but not in SENDING or FINISHING state (%d)\n", int(transfer.state));
 		return;
 	}
 
@@ -813,16 +878,20 @@ static void _handle_FT1_DATA_ACK(
 
 	transfer.time_since_activity = 0.f;
 
+	std::vector<LEDBAT::SeqIDType> seqs;
 	while (curser < length) {
 		uint16_t seq_id = data[curser++];
 		seq_id |= data[curser++] << (1*8);
 
+		seqs.push_back({transfer_id, seq_id});
 		transfer.ssb.erase(seq_id);
 	}
+	peer.cca.onAck(seqs);
 
-	if (transfer.file_size == transfer.file_size_current) {
+	// delete if all packets acked
+	if (transfer.file_size == transfer.file_size_current && transfer.ssb.size() == 0) {
 		fprintf(stderr, "FT: %d done\n", transfer_id);
-		peer.send_transfers[transfer_id] = std::nullopt;
+		peer.send_transfers[transfer_id].reset();
 	}
 }
 

ngc_ft1.h

@@ -17,19 +17,25 @@ extern "C" {
 typedef struct NGC_FT1 NGC_FT1;
 
 struct NGC_FT1_options {
-	// TODO: expose some parameters
-	int tmp;
+	// TODO
+	size_t acks_per_packet; // 3
+
+	float init_retry_timeout_after; // 10sec
+
+	//float sending_resend_without_ack_after; // 5sec
+	float sending_give_up_after; // 30sec
 };
 
-// uint16_t ?
+// uint32_t - same as tox friend ft
 // ffs c does not allow types
-typedef enum NGC_FT1_file_kind /*: uint8_t*/ {
-	//INVALID = 0u,
+typedef enum NGC_FT1_file_kind /*: uint32_t*/ {
+	//INVALID = 0u, // DATA?
 
 	// id:
 	// group (implicit)
 	// peer pub key + msg_id
 	NGC_HS1_MESSAGE_BY_ID = 1u, // history sync PoC 1
+	// TODO: oops, 1 should be avatar v1
 
 	// id: TOX_FILE_ID_LENGTH (32) bytes
 	// this is basically and id and probably not a hash, like the tox friend api
@@ -41,32 +47,54 @@ typedef enum NGC_FT1_file_kind /*: uint8_t*/ {
 	// draft: (for single file)
 	//   - 256 bytes | filename
 	//   - 8bytes | file size
+	//   - 4bytes | chunk size
+	//   - array of chunk hashes (ids) [
+	//     - SHA1 bytes (20)
+	//   - ]
+	HASH_SHA1_INFO,
+	// draft: (for single file) v2
+	//   - c-string | filename
+	//   - 8bytes | file size
+	//   - 4bytes | chunk size
+	//   - array of chunk hashes (ids) [
+	//     - SHA1 bytes (20)
+	//   - ]
+	HASH_SHA1_INFO2,
+	// draft: multiple files
+	//   - 4bytes | number of filenames
+	//   - array of filenames (variable length c-strings) [
+	//     - c-string | filename (including path and '/' as dir seperator)
+	//   - ]
+	//   - 256 bytes | filename
+	//   - 8bytes | file size
 	//   - fixed chunk size of 4kb
 	//   - array of chunk hashes (ids) [
 	//     - SHAX bytes
 	//   - ]
-	HASH_SHA1_INFO,
-	HASH_SHA2_INFO,
+	HASH_SHA1_INFO3,
+	HASH_SHA2_INFO, // hm?
 
 	// id: hash of the content
 	// TODO: fixed chunk size or variable (defined in info)
-	// if "variable" sized, it can be aliased with TORRENT_VX_CHUNK in the implementation
+	// if "variable" sized, it can be aliased with TORRENT_V1_CHUNK in the implementation
 	HASH_SHA1_CHUNK,
 	HASH_SHA2_CHUNK,
 
-	// :)
-	// draft for fun and profit
-	// TODO: should we even support v1?
 	// TODO: design the same thing again for tox? (msg_pack instead of bencode?)
 	// id: infohash
 	TORRENT_V1_METAINFO,
 	// id: sha1
-	TORRENT_V1_CHUNK, // alias with SHA1_CHUNK?
+	TORRENT_V1_PIECE, // alias with SHA1_CHUNK?
 
 	// id: infohash
-	TORRENT_V2_METAINFO, // meta info is kind of more complicated than that <.<
+	// in v2, metainfo contains only the root hashes of the merkletree(s)
+	TORRENT_V2_METAINFO,
+	// id: root hash
+	// contains all the leaf hashes for a file root hash
+	TORRENT_V2_FILE_HASHES,
 	// id: sha256
-	TORRENT_V2_CHUNK,
+	// always of size 16KiB, except if last piece in file
+	TORRENT_V2_PIECE,
 } NGC_FT1_file_kind;
 
 // ========== init / kill ==========
@@ -76,7 +104,8 @@ bool NGC_FT1_register_ext(NGC_FT1* ngc_ft1_ctx, NGC_EXT_CTX* ngc_ext_ctx);
 void NGC_FT1_kill(NGC_FT1* ngc_ft1_ctx);
 
 // ========== iterate ==========
-void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx);
+// time_delta in seconds
+void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx, float time_delta);
 
 // TODO: announce
 // ========== request ==========
@@ -85,7 +114,7 @@ void NGC_FT1_iterate(Tox *tox, NGC_FT1* ngc_ft1_ctx);
 void NGC_FT1_send_request_private(
 	Tox *tox, NGC_FT1* ngc_ft1_ctx,
 	uint32_t group_number, uint32_t peer_number,
-	NGC_FT1_file_kind file_kind,
+	uint32_t file_kind,
 	const uint8_t* file_id, size_t file_id_size
 );
 
@@ -98,7 +127,7 @@ typedef void NGC_FT1_recv_request_cb(
 
 void NGC_FT1_register_callback_recv_request(
 	NGC_FT1* ngc_ft1_ctx,
-	NGC_FT1_file_kind file_kind,
+	uint32_t file_kind,
 	NGC_FT1_recv_request_cb* callback,
 	void* user_data
 );
@@ -109,7 +138,7 @@ void NGC_FT1_register_callback_recv_request(
 bool NGC_FT1_send_init_private(
 	Tox *tox, NGC_FT1* ngc_ft1_ctx,
 	uint32_t group_number, uint32_t peer_number,
-	NGC_FT1_file_kind file_kind,
+	uint32_t file_kind,
 	const uint8_t* file_id, size_t file_id_size,
 	size_t file_size,
 	uint8_t* transfer_id
@@ -127,7 +156,7 @@ typedef bool NGC_FT1_recv_init_cb(
 
 void NGC_FT1_register_callback_recv_init(
 	NGC_FT1* ngc_ft1_ctx,
-	NGC_FT1_file_kind file_kind,
+	uint32_t file_kind,
 	NGC_FT1_recv_init_cb* callback,
 	void* user_data
 );
@@ -147,7 +176,7 @@ typedef void NGC_FT1_recv_data_cb(
 
 void NGC_FT1_register_callback_recv_data(
 	NGC_FT1* ngc_ft1_ctx,
-	NGC_FT1_file_kind file_kind,
+	uint32_t file_kind,
 	NGC_FT1_recv_data_cb* callback,
 	void* user_data
 );
@@ -166,7 +195,7 @@ typedef void NGC_FT1_send_data_cb(
 
 void NGC_FT1_register_callback_send_data(
 	NGC_FT1* ngc_ft1_ctx,
-	NGC_FT1_file_kind file_kind,
+	uint32_t file_kind,
 	NGC_FT1_send_data_cb* callback,
 	void* user_data
 );