Send the announces that outbound() queues

src/microReticulum/Interface.cpp

@@ -16,9 +16,11 @@
 
 #include "Identity.h"
 #include "Transport.h"
+#include "Utilities/OS.h"
 
 using namespace RNS;
 using namespace RNS::Type::Interface;
+using namespace RNS::Utilities;
 
 /*static*/ uint8_t Interface::DISCOVER_PATHS_FOR = MODE_ACCESS_POINT | MODE_GATEWAY | MODE_ROAMING;
 
@@ -86,51 +88,68 @@ void Interface::handle_incoming(const Bytes& data) {
     }
 }
 
+// Send an announce that Transport::outbound had to queue, as often as the
+// interface's announce cap allows one out.
+//
+// RNS arms a timer for the cap's wait and sends the next announce when it
+// fires. There are no timers here, so Transport::jobs calls this on every
+// interface and the cap is kept by _announce_allowed_at instead: the interval
+// between calls decides only how promptly a queue drains, never how fast
+// announces leave. One per call is what the timer does per firing, and is far
+// more than the arrival rate a queue forms from.
+//
+// While this had no body nothing ever drained the queue, and outbound()
+// refuses to transmit an announce onto an interface that has any queued: the
+// first announce an interface ever queued stopped it forwarding announces for
+// the rest of the uptime, silently and with no way back short of a reboot.
 void Interface::process_announce_queue() {
-/*
-	if not hasattr(self, "announce_cap"):
-		self.announce_cap = RNS.Reticulum.ANNOUNCE_CAP
-
-	if hasattr(self, "announce_queue"):
-		try:
-			now = time.time()
-			stale = []
-			for a in self.announce_queue:
-				if now > a["time"]+RNS.Reticulum.QUEUED_ANNOUNCE_LIFE:
-					stale.append(a)
-
-			for s in stale:
-				if s in self.announce_queue:
-					self.announce_queue.remove(s)
-
-			if len(self.announce_queue) > 0:
-				min_hops = min(entry["hops"] for entry in self.announce_queue)
-				entries = list(filter(lambda e: e["hops"] == min_hops, self.announce_queue))
-				entries.sort(key=lambda e: e["time"])
-				selected = entries[0]
-
-				double now = OS::time();
-				uint32_t wait_time = 0;
-				if (_impl->_bitrate > 0 && _impl->_announce_cap > 0) {
-					uint32_t tx_time = (len(selected["raw"])*8) / _impl->_bitrate;
-					wait_time = (tx_time / _impl->_announce_cap);
-				}
-				_impl->_announce_allowed_at = now + wait_time;
-
-				self.on_outgoing(selected["raw"])
-
-				if selected in self.announce_queue:
-					self.announce_queue.remove(selected)
-
-				if len(self.announce_queue) > 0:
-					timer = threading.Timer(wait_time, self.process_announce_queue)
-					timer.start()
-
-		except Exception as e:
-			self.announce_queue = []
-			RNS.log("Error while processing announce queue on "+str(self)+". The contained exception was: "+str(e), RNS.LOG_ERROR)
-			RNS.log("The announce queue for this interface has been cleared.", RNS.LOG_ERROR)
-*/
+	assert(_impl);
+	if (_impl->_announce_queue.empty()) return;
+
+	try {
+		double now = OS::time();
+
+		// An announce that has been waiting this long is not worth the airtime
+		// any more: whoever sent it has almost certainly announced again since.
+		size_t held = _impl->_announce_queue.size();
+		_impl->_announce_queue.remove_if([now](const AnnounceEntry& entry) {
+			return now > entry._time + Type::Reticulum::QUEUED_ANNOUNCE_LIFE;
+		});
+		if (_impl->_announce_queue.size() < held) {
+			DEBUGF("Dropped %u stale queued announce(s) on %s",
+				(unsigned)(held - _impl->_announce_queue.size()), toString().c_str());
+		}
+		if (_impl->_announce_queue.empty() || now < _impl->_announce_allowed_at) return;
+
+		// Fewest hops first, and among those the one that has waited longest,
+		// so a queue draining slowly still carries the nearest path onwards
+		// first and nothing in it can be starved by later arrivals.
+		auto selected = _impl->_announce_queue.begin();
+		for (auto entry = _impl->_announce_queue.begin(); entry != _impl->_announce_queue.end(); ++entry) {
+			if (entry->_hops < selected->_hops ||
+				(entry->_hops == selected->_hops && entry->_time < selected->_time)) {
+				selected = entry;
+			}
+		}
+
+		double wait_time = announce_wait_time(selected->_raw.size());
+		_impl->_announce_allowed_at = now + wait_time;
+
+		Bytes raw(selected->_raw);
+		_impl->_announce_queue.erase(selected);
+		DEBUGF("Sending queued announce on %s, %u still queued, next allowed in %.1f s",
+			toString().c_str(), (unsigned)_impl->_announce_queue.size(), wait_time);
+
+		if (Transport::transmit(*this, raw)) {
+			sent_announce();
+		}
+	}
+	catch (const std::exception& e) {
+		_impl->_announce_queue.clear();
+		ERRORF("Error while processing the announce queue on %s. The contained exception was: %s",
+			toString().c_str(), e.what());
+		ERROR("The announce queue for this interface has been cleared.");
+	}
 }
 
 /*

src/microReticulum/Interface.h

@@ -210,6 +210,22 @@ namespace RNS {
 		inline const Bytes get_hash() const { assert(_impl); return _impl->get_hash(); }
 		void process_announce_queue();
 
+		// How long the announce cap makes this interface wait after sending an
+		// announce of this size: the packet's airtime divided by the share of
+		// the link announces are allowed to take. Both the callers that need it
+		// -- Transport::outbound when it lets an announce through, and
+		// process_announce_queue when it sends a queued one -- ask here, so the
+		// two cannot disagree about when the next announce is due. Answered in
+		// seconds, in floating point: computed in integers, as one caller used
+		// to, every wait below one second rounds to none at all, which is every
+		// wait an interface faster than a slow LoRa channel ever has.
+		inline double announce_wait_time(size_t tx_size) const {
+			assert(_impl);
+			if (_impl->_bitrate == 0 || _impl->_announce_cap <= 0.0) return 0.0;
+			double tx_time = (double)(tx_size * 8) / (double)_impl->_bitrate;
+			return tx_time / (double)_impl->_announce_cap;
+		}
+
 		// CBA ACCUMULATES
 		inline void add_announce(AnnounceEntry& entry) { assert(_impl); _impl->_announce_queue.push_back(entry); }
 

src/microReticulum/Transport.cpp

@@ -1007,6 +1007,22 @@ TRACEF("path_request_conditions=%u", path_request_conditions);
             // Run interface-related jobs
 			if (OS::time() > (_interface_last_jobs + _interface_jobs_interval)) {
 				prioritize_interfaces();
+				// Announces that outbound() had to queue are sent from here.
+				// RNS arms a timer per interface for this; this port has none,
+				// so the queues are polled and each interface's own announce
+				// cap decides whether anything actually goes out. Nothing
+				// called process_announce_queue() at all before, and outbound()
+				// will not transmit an announce while an interface has one
+				// queued -- so an interface that queued a single announce
+				// forwarded no further announce for the rest of the uptime.
+				try {
+					for (auto& interface : _interfaces) {
+						interface.process_announce_queue();
+					}
+				}
+				catch (const std::exception& e) {
+					ERRORF("Error while processing queued per-interface announces: %s", e.what());
+				}
 				// TODO
 /*
 				try {
@@ -1393,12 +1409,7 @@ TRACEF("path_request_conditions=%u", path_request_conditions);
 
 								bool queued_announces = (interface.announce_queue().size() > 0);
 								if (!queued_announces && outbound_time > interface.announce_allowed_at()) {
-									uint16_t wait_time = 0;
-									if (interface.bitrate() > 0 && interface.announce_cap() > 0) {
-										uint16_t tx_time = (packet.raw().size() * 8) / interface.bitrate();
-										wait_time = (tx_time / interface.announce_cap());
-									}
-									interface.announce_allowed_at(outbound_time + wait_time);
+									interface.announce_allowed_at(outbound_time + interface.announce_wait_time(packet.raw().size()));
 								}
 								else {
 									should_transmit = false;
@@ -3784,12 +3795,7 @@ will announce it.
 				return;
 			}
 			else {
-				//p tx_time   = ((len(path_request_data)+RNS.Reticulum.HEADER_MINSIZE)*8) / on_interface.bitrate
-				uint32_t wait_time = 0;
-				if ( on_interface.bitrate() > 0 && on_interface.announce_cap() > 0) {
-					uint32_t tx_time = ((path_request_data.size() + Type::Reticulum::HEADER_MINSIZE)*8) / on_interface.bitrate();
-					wait_time = (tx_time / on_interface.announce_cap());
-				}
+				double wait_time = on_interface.announce_wait_time(path_request_data.size() + Type::Reticulum::HEADER_MINSIZE);
 				const_cast<Interface&>(on_interface).announce_allowed_at(now + wait_time);
 			}
 		}

test/test_transport/test_transport.cpp

@@ -631,6 +631,165 @@ void test_receipt_timeout_handler_capture() {
 void setUp(void) {}
 void tearDown(void) {}
 
+// ============================================================================
+// Announce queue
+// ============================================================================
+
+// Counts what it is asked to send and keeps the last of it, so a test can say
+// which queued announce went out and when. Bitrate and announce cap are what
+// decide the wait between two announces, so each test sets the pair its case
+// is about.
+class QueueInterface : public RNS::InterfaceImpl {
+public:
+	QueueInterface(uint32_t bitrate, float cap, const char* name = "QueueInterface")
+		: RNS::InterfaceImpl(name) {
+		_OUT = true;
+		_IN = false;
+		_bitrate = bitrate;
+		_announce_cap = cap;
+	}
+	virtual ~QueueInterface() { _name = "(deleted)"; }
+	virtual bool send_outgoing(const RNS::Bytes& data) {
+		_sent++;
+		_last = data;
+		return true;
+	}
+public:
+	size_t _sent = 0;
+	RNS::Bytes _last;
+};
+
+// A queued announce of a given age and hop count, its raw bytes filled with a
+// marker so a test can tell which one was sent.
+static RNS::AnnounceEntry test_announce(uint8_t hops, double time, uint8_t marker) {
+	uint8_t raw[100];
+	memset(raw, marker, sizeof(raw));
+	uint8_t destination[1] = { marker };
+	return RNS::AnnounceEntry(RNS::Bytes(destination, sizeof(destination)), time, hops, 0,
+		RNS::Bytes(raw, sizeof(raw)));
+}
+
+// The fewest hops leaves first, and the cap holds the rest back. Both halves
+// matter: without the first the queue is a queue in name only, and without the
+// second an interface can be talked into spending its whole link on announces.
+void test_announce_queue_sends_fewest_hops_first() {
+
+	printf("test_announce_queue_sends_fewest_hops_first: BEGIN\n");
+
+	// 1200 bits/s at a 2 % cap: a 100-byte announce owes about 33 s before the
+	// next one may go. Computed in integers, as this used to be, the whole wait
+	// truncates to nothing and the cap never engages at all.
+	QueueInterface* impl = new QueueInterface(1200, 0.02f);
+	RNS::Interface iface(impl);
+
+	double now = RNS::Utilities::OS::time();
+	RNS::AnnounceEntry far(test_announce(3, now, 'f'));
+	RNS::AnnounceEntry near(test_announce(1, now, 'n'));
+	RNS::AnnounceEntry mid(test_announce(2, now, 'm'));
+	iface.add_announce(far);
+	iface.add_announce(near);
+	iface.add_announce(mid);
+	TEST_ASSERT_EQUAL_UINT32(3, iface.announce_queue().size());
+
+	iface.process_announce_queue();
+
+	TEST_ASSERT_EQUAL_UINT32(1, impl->_sent);
+	TEST_ASSERT_EQUAL_UINT32(2, iface.announce_queue().size());
+	TEST_ASSERT_EQUAL_UINT8('n', impl->_last.data()[0]);
+	// About 33 s of it: 800 bits at 1200 bits/s is two thirds of a second of
+	// airtime, and a 2 % cap is fifty times that. Unity's double assertions
+	// are compiled out here, so the comparison is a plain one.
+	double wait = iface.announce_allowed_at() - now;
+	TEST_ASSERT_TRUE(wait > 30.0 && wait < 36.0);
+
+	// Called again straight away, the cap has not elapsed and nothing else goes.
+	iface.process_announce_queue();
+	TEST_ASSERT_EQUAL_UINT32(1, impl->_sent);
+	TEST_ASSERT_EQUAL_UINT32(2, iface.announce_queue().size());
+
+	printf("test_announce_queue_sends_fewest_hops_first: END\n");
+}
+
+// The queue drains once the cap's wait has passed. Nothing drained it at all
+// before -- and because Transport::outbound refuses to transmit an announce on
+// an interface that has one queued, a single queued announce used to stop that
+// interface forwarding announces for the rest of the uptime.
+void test_announce_queue_drains_when_the_cap_allows() {
+
+	printf("test_announce_queue_drains_when_the_cap_allows: BEGIN\n");
+
+	// 10 Mbit/s at 2 %: a 100-byte announce owes 4 ms.
+	QueueInterface* impl = new QueueInterface(10000000, 0.02f);
+	RNS::Interface iface(impl);
+
+	double now = RNS::Utilities::OS::time();
+	RNS::AnnounceEntry first(test_announce(1, now, 'a'));
+	RNS::AnnounceEntry second(test_announce(1, now + 0.001, 'b'));
+	iface.add_announce(first);
+	iface.add_announce(second);
+
+	iface.process_announce_queue();
+	TEST_ASSERT_EQUAL_UINT32(1, impl->_sent);
+	TEST_ASSERT_EQUAL_UINT8('a', impl->_last.data()[0]);      // same hops: oldest first
+
+	RNS::Utilities::OS::sleep(0.02);
+	iface.process_announce_queue();
+	TEST_ASSERT_EQUAL_UINT32(2, impl->_sent);
+	TEST_ASSERT_EQUAL_UINT8('b', impl->_last.data()[0]);
+	TEST_ASSERT_EQUAL_UINT32(0, iface.announce_queue().size());
+
+	// An empty queue is not an error, and asks nothing of the interface.
+	iface.process_announce_queue();
+	TEST_ASSERT_EQUAL_UINT32(2, impl->_sent);
+
+	printf("test_announce_queue_drains_when_the_cap_allows: END\n");
+}
+
+// An announce that has waited longer than a queued announce may live is not
+// worth the airtime: whoever sent it has announced again several times since.
+void test_stale_queued_announces_are_dropped() {
+
+	printf("test_stale_queued_announces_are_dropped: BEGIN\n");
+
+	QueueInterface* impl = new QueueInterface(10000000, 0.02f);
+	RNS::Interface iface(impl);
+
+	double now = RNS::Utilities::OS::time();
+	RNS::AnnounceEntry stale(test_announce(1, now - (double)RNS::Type::Reticulum::QUEUED_ANNOUNCE_LIFE - 1, 's'));
+	RNS::AnnounceEntry fresh(test_announce(1, now, 'f'));
+	iface.add_announce(stale);
+	iface.add_announce(fresh);
+
+	iface.process_announce_queue();
+
+	TEST_ASSERT_EQUAL_UINT32(1, impl->_sent);
+	TEST_ASSERT_EQUAL_UINT8('f', impl->_last.data()[0]);
+	TEST_ASSERT_EQUAL_UINT32(0, iface.announce_queue().size());
+
+	printf("test_stale_queued_announces_are_dropped: END\n");
+}
+
+// An interface that has not said how fast it is owes no wait, and must still
+// drain rather than sit on the queue forever.
+void test_announce_queue_drains_without_a_bitrate() {
+
+	printf("test_announce_queue_drains_without_a_bitrate: BEGIN\n");
+
+	QueueInterface* impl = new QueueInterface(0, 0.02f);
+	RNS::Interface iface(impl);
+
+	double now = RNS::Utilities::OS::time();
+	RNS::AnnounceEntry only(test_announce(1, now, 'o'));
+	iface.add_announce(only);
+
+	iface.process_announce_queue();
+
+	TEST_ASSERT_EQUAL_UINT32(1, impl->_sent);
+	TEST_ASSERT_EQUAL_UINT32(0, iface.announce_queue().size());
+
+	printf("test_announce_queue_drains_without_a_bitrate: END\n");
+}
+
 int runUnityTests(void) {
 	UNITY_BEGIN();
 
@@ -658,6 +817,12 @@ int runUnityTests(void) {
 */
 	RUN_TEST(test_prioritize_interfaces);
 	RUN_TEST(test_incoming_announce_over_limit);
+
+	// Announce queue
+	RUN_TEST(test_announce_queue_sends_fewest_hops_first);
+	RUN_TEST(test_announce_queue_drains_when_the_cap_allows);
+	RUN_TEST(test_stale_queued_announces_are_dropped);
+	RUN_TEST(test_announce_queue_drains_without_a_bitrate);
 	//RUN_TEST(test_incoming_announce_stress);
 
 #if RNS_NEIGHBOR_PROBING