refactor sending transfers

always do timeouts and resending first
then roundrobin over sending new data

solanaceae/ngc_ft1/flow_only.cpp

@@ -170,7 +170,7 @@ void FlowOnly::onAck(std::vector<SeqIDType> seqs) {
 			// find first non ignore, it should be the expected
 			auto first_it = std::find_if_not(_in_flight.cbegin(), _in_flight.cend(), [](const auto& v) -> bool { return v.ignore; });
 
-			if (first_it != _in_flight.cend() && it != first_it) {
+			if (first_it != _in_flight.cend() && it != first_it && !it->ignore) {
 				// not next expected seq -> skip detected
 
 				_consecutive_events++;

solanaceae/ngc_ft1/ngcft1.cpp

@@ -15,20 +15,19 @@
 #include <cassert>
 #include <vector>
 
-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, int64_t& can_packet_size) {
+void NGCFT1::updateSendTransferPhase1(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer, size_t idx, std::set<CCAI::SeqIDType>& timeouts_set, int64_t& can_packet_size) {
+	using State = Group::Peer::SendTransfer::State;
 	auto& tf_opt = peer.send_transfers.at(idx);
 	assert(tf_opt.has_value());
 	auto& tf = tf_opt.value();
 
 	tf.time_since_activity += time_delta;
 
-	switch (tf.state) {
-		using State = Group::Peer::SendTransfer::State;
-		case State::INIT_SENT:
+	if (tf.state == State::INIT_SENT) {
 		if (tf.time_since_activity >= init_retry_timeout_after) {
 			if (tf.inits_sent >= 3) {
 				// delete, timed out 3 times
-					std::cerr << "NGCFT1 warning: ft init timed out, deleting\n";
+				std::cerr << "NGCFT1 warning: sending ft init timed out, deleting\n";
 				dispatch(
 					NGCFT1_Event::send_done,
 					Events::NGCFT1_send_done{
@@ -39,55 +38,22 @@ void NGCFT1::updateSendTransfer(float time_delta, uint32_t group_number, uint32_
 				tf_opt.reset();
 			} else {
 				// timed out, resend
-					std::cerr << "NGCFT1 warning: ft init timed out, resending\n";
+				std::cerr << "NGCFT1 warning: sending ft init timed out, resending\n";
 				_neep.send_ft1_init(group_number, peer_number, tf.file_kind, tf.file_size, idx, tf.file_id.data(), tf.file_id.size());
 				tf.inits_sent++;
 				tf.time_since_activity = 0.f;
 			}
 		}
-			break;
-		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) {
-				if (timeouts_set.count({idx, id})) {
-					if (can_packet_size >= int64_t(data.size())) {
-						_neep.send_ft1_data(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});
-						can_packet_size -= data.size();
-					} else {
-#if 0 // too spammy
-						std::cerr << "NGCFT1 warning: no space to resend timedout\n";
-#endif
-					}
-				}
-			});
-			if (tf.time_since_activity >= (sending_give_up_after * peer.active_send_transfers)) {
-				// no ack after 30sec, close ft
-				std::cerr << "NGCFT1 warning: sending ft finishing timed out, deleting\n";
-				dispatch(
-					NGCFT1_Event::send_done,
-					Events::NGCFT1_send_done{
-						group_number, peer_number,
-						static_cast<uint8_t>(idx),
-					}
-				);
-
-				// 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;
-		case State::SENDING: {
-				// first handle overall timeout (could otherwise do resends directly before, which is useless)
+		return;
+	} else if (tf.state == State::FINISHING || tf.state == State::SENDING) {
 		// timeout increases with active transfers (otherwise we could starve them)
 		if (tf.time_since_activity >= (sending_give_up_after * peer.active_send_transfers)) {
-					// no ack after 30sec, close ft
+			// no ack after Xsec, close ft
+			if (tf.state == State::FINISHING) {
+				std::cerr << "NGCFT1 warning: sending ft finishing timed out, deleting\n";
+			} else {
 				std::cerr << "NGCFT1 warning: sending ft in progress timed out, deleting (ifc:" << peer.cca->inFlightCount() << ")\n";
+			}
 			dispatch(
 				NGCFT1_Event::send_done,
 				Events::NGCFT1_send_done{
@@ -99,25 +65,60 @@ void NGCFT1::updateSendTransfer(float time_delta, uint32_t group_number, uint32_
 			// 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
+		}
 		return;
 	}
 
-				// do resends
+	// do send buffer and resending
 	tf.ssb.for_each(time_delta, [&](uint16_t id, const std::vector<uint8_t>& data, float& time_since_activity) {
-					if (can_packet_size >= int64_t(data.size()) && time_since_activity >= peer.cca->getCurrentDelay() && timeouts_set.count({idx, id})) {
-						// TODO: can fail
-						_neep.send_ft1_data(group_number, peer_number, idx, id, data.data(), data.size());
-						peer.cca->onLoss({idx, id}, false);
+		time_since_activity += time_delta;
+
+		if (tf.state != State::FINISHING && tf.state != State::SENDING) {
+			return;
+		}
+
+		if (
+			time_since_activity >= peer.cca->getCurrentDelay() && // TODO: use OR instead?
+			timeouts_set.count({idx, id})
+		) {
+			if (can_packet_size >= int64_t(data.size() /*+ peer.cca->SEGMENT_OVERHEAD*/)) {
+				if (_neep.send_ft1_data(group_number, peer_number, idx, id, data.data(), data.size())) {
+					if (!peer.cca->onLoss({idx, id}, false)) { // might not be in cca
+						peer.cca->onSent({idx, id}, data.size());
+					}
+
 					time_since_activity = 0.f;
-						timeouts_set.erase({idx, id});
 					can_packet_size -= data.size();
+				} else {
+					std::cerr << "NGCFT1 warning: failed to re-send packet (send queue full?)\n";
+
+					// signal ce (we did not call onLoss()
+					peer.cca->onCongestion();
+					can_packet_size = 0;
+				}
+#if 0
+			} else {
+				std::cerr << "NGCFT1 warning: no space to resend timed-out\n";
+#endif
+			}
+
+			timeouts_set.erase({idx, id});
 		}
 	});
+}
+
+void NGCFT1::updateSendTransferPhase2(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer, size_t idx, int64_t& can_packet_size) {
+	using State = Group::Peer::SendTransfer::State;
+	auto& tf_opt = peer.send_transfers.at(idx);
+	assert(tf_opt.has_value());
+	auto& tf = tf_opt.value();
+
+	if (tf.state != State::SENDING) {
+		return;
+	}
 
 	// if chunks in flight < window size (2)
 	while (can_packet_size > 0 && tf.file_size > 0) {
@@ -152,7 +153,7 @@ void NGCFT1::updateSendTransfer(float time_delta, uint32_t group_number, uint32_
 		if (sent) {
 			peer.cca->onSent({idx, seq_id}, chunk_size);
 		} else {
-						std::cerr << "NGCFT1: failed to send packet (queue full?) --------------\n";
+			std::cerr << "NGCFT1 warn: failed to send packet (send queue full?)\n";
 			peer.cca->onCongestion();
 			can_packet_size = 0;
 		}
@@ -160,14 +161,6 @@ void NGCFT1::updateSendTransfer(float time_delta, uint32_t group_number, uint32_
 		tf.file_size_current += chunk_size;
 		can_packet_size -= chunk_size;
 	}
-			}
-			break;
-		default: // invalid state, delete
-			std::cerr << "NGCFT1 error: ft in invalid state, deleting\n";
-			assert(false && "ft in invalid state");
-			tf_opt.reset();
-			return;
-	}
 }
 
 bool NGCFT1::iteratePeer(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer) {
@@ -201,14 +194,17 @@ bool NGCFT1::iteratePeer(float time_delta, uint32_t group_number, uint32_t peer_
 
 		int64_t can_packet_size {peer.cca->canSend(time_delta)}; // might get more space while iterating (time)
 
-		// get number current running transfers TODO: improve
+		// resend and get number current running transfers
 		peer.active_send_transfers = 0;
-		for (const auto& it : peer.send_transfers) {
-			if (it.has_value()) {
-				peer.active_send_transfers++;
+		for (size_t idx = 0; idx < peer.send_transfers.size(); idx++) {
+			if (!peer.send_transfers.at(idx).has_value()) {
+				continue;
 			}
+			peer.active_send_transfers++;
+			updateSendTransferPhase1(time_delta, group_number, peer_number, peer, idx, timeouts_set, can_packet_size);
 		}
 
+		if (can_packet_size > 0) {
 			// change iterate start position to not starve transfers in the back
 			size_t iterated_count = 0;
 			bool last_send_found = false;
@@ -220,7 +216,8 @@ bool NGCFT1::iteratePeer(float time_delta, uint32_t group_number, uint32_t peer_
 						peer.next_send_transfer_send_idx = idx;
 						last_send_found = true; // only set once
 					}
-				updateSendTransfer(time_delta, group_number, peer_number, peer, idx, timeouts_set, can_packet_size);
+					updateSendTransferPhase2(time_delta, group_number, peer_number, peer, idx, can_packet_size);
+				}
 			}
 		}
 	}

solanaceae/ngc_ft1/ngcft1.hpp

@@ -209,7 +209,11 @@ class NGCFT1 : public ToxEventI, public NGCEXTEventI, public NGCFT1EventProvider
 	std::map<uint32_t, Group> groups;
 
 	protected:
-		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, int64_t& can_packet_size);
+		// general update with timeouts and resending
+		void updateSendTransferPhase1(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer, size_t idx, std::set<CCAI::SeqIDType>& timeouts_set, int64_t& can_packet_size);
+		// does sending new data
+		void updateSendTransferPhase2(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer, size_t idx, int64_t& can_packet_size);
+
 		bool iteratePeer(float time_delta, uint32_t group_number, uint32_t peer_number, Group::Peer& peer);
 
 		const CCAI* getPeerCCA(uint32_t group_number, uint32_t peer_number) const;

solanaceae/ngc_ft1/snd_buf.hpp

@@ -26,7 +26,6 @@ struct SendSequenceBuffer {
 	template<typename FN>
 	void for_each(float time_delta, FN&& fn) {
 		for (auto& [id, entry] : entries) {
-			entry.time_since_activity += time_delta;
 			fn(id, entry.data, entry.time_since_activity);
 		}
 	}