diff --git a/src/microReticulum/Interface.cpp b/src/microReticulum/Interface.cpp index 46cd09d5..940b0bca 100644 --- a/src/microReticulum/Interface.cpp +++ b/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."); + } } /* diff --git a/src/microReticulum/Interface.h b/src/microReticulum/Interface.h index 0d939b9b..099a84ca 100644 --- a/src/microReticulum/Interface.h +++ b/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); } diff --git a/src/microReticulum/Transport.cpp b/src/microReticulum/Transport.cpp index 89b5238b..6fc094b8 100644 --- a/src/microReticulum/Transport.cpp +++ b/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(on_interface).announce_allowed_at(now + wait_time); } } diff --git a/test/test_transport/test_transport.cpp b/test/test_transport/test_transport.cpp index 9d956433..b2df1612 100644 --- a/test/test_transport/test_transport.cpp +++ b/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