Periodically reset window if not configured to broadcast stats

Author: erayd

src/modules/ReplayModule.cpp

@@ -0,0 +1,308 @@
+#include "ReplayModule.h"
+#include "memGet.h"
+#include "modules/RoutingStatsModule.h"
+
+ReplayModule *replayModule{};
+
+/**
+ * Constructor
+ */
+ReplayModule::ReplayModule() : SinglePortModule("Replay", meshtastic_PortNum_REPLAY_APP), NotifiedWorkerThread("Replay")
+{
+    notify((uint32_t)ReplayNotification::INIT, true);
+}
+
+/**
+ * Adopt a packet into the replay buffer
+ */
+ReplayPacket *ReplayModule::adoptPacket(const meshtastic_MeshPacket *p, bool from_tx)
+{
+    ReplayPacket *r = learnPacket(PACKET_HASH(p->from, p->id), getPriority(p));
+    if (from_tx)
+        r->last_tx_millis = millis();
+    else
+        r->last_rx_millis = millis();
+
+    if (!r->cache) {
+        r->cache = packetCache.cache(p, false);
+        if (r->cache) {
+            cache_packets++;
+            cache_bytes += sizeof(*r->cache) + r->cache->payload_len;
+            dirty.set(r->idx);
+            dirty_fast.set(r->idx);
+            routingStats->logEvent(RoutingEvent::REPLAY_CACHE_PACKETS, NULL, cache_packets);
+            routingStats->logEvent(RoutingEvent::REPLAY_CACHE_BYTES, NULL, cache_bytes);
+            pruneCache();
+            LOG_DEBUG("Replay: Added packet to cache (packets=%lu bytes=%lu hash=0x%04x from=0x%08x id=0x%08x)", cache_packets,
+                      cache_bytes, r->hash, p->from, p->id);
+        }
+    }
+    r->missed = false;
+
+    return r;
+}
+
+/**
+ * Get the replay priority for a packet
+ */
+ReplayPriority ReplayModule::getPriority(const meshtastic_MeshPacket *p)
+{
+    ReplayPriority prio = meshtastic_Config_ReplayConfig_ReplayPriority_NORMAL;
+    if (!p)
+        return prio; // This isn't a packet
+
+    if (p->which_payload_variant == meshtastic_MeshPacket_decoded_tag) {
+
+        // Adjustments for decrypted packets
+        if (p->decoded.portnum >= 64)
+            prio = meshtastic_Config_ReplayConfig_ReplayPriority_LOW; // Non-core apps are assumed low priority by default
+        // App-specific adjustments
+        switch (p->decoded.portnum) {
+        case meshtastic_PortNum_UNKNOWN_APP:          // Opaque app traffic, we don't understand it
+        case meshtastic_PortNum_POSITION_APP:         // Position updates are frequent and automatic
+        case meshtastic_PortNum_AUDIO_APP:            // Large payload, very inefficient, demote it to bottom of the heap
+        case meshtastic_PortNum_IP_TUNNEL_APP:        // Potentially large source of traffic, it's abusive, demote it
+        case meshtastic_PortNum_STORE_FORWARD_APP:    // Potentially large source of traffic, demote it
+        case meshtastic_PortNum_RANGE_TEST_APP:       // These are zero-hop anyway
+        case meshtastic_PortNum_TELEMETRY_APP:        // Frequent and automatic, demote it
+        case meshtastic_PortNum_NEIGHBORINFO_APP:     // Automatic and repeated, reliability via redundancy rather than replay
+        case meshtastic_PortNum_RETICULUM_TUNNEL_APP: // Potentially large source of foreign traffic, demote it
+        case meshtastic_PortNum_CAYENNE_APP:          // LoRaWAN sensor data
+        case _meshtastic_PortNum_MAX:                 // This isn't a real app, somebody probably screwed up, so deprioritise it
+            prio = meshtastic_Config_ReplayConfig_ReplayPriority_BACKGROUND;
+            break;
+        case meshtastic_PortNum_NODEINFO_APP:   // Nodeinfo is repeated often and likely cached, so give way to other stuff:
+        case meshtastic_PortNum_SERIAL_APP:     // Chatty, not as bad as IP
+        case meshtastic_PortNum_TRACEROUTE_APP: // Meshsense gets abusive with this, demote despite potentially
+                                                // human-initiated
+        case meshtastic_PortNum_ATAK_PLUGIN:    // Can get very chatty, degrade gracefully by being more eager to drop this
+        case meshtastic_PortNum_ATAK_FORWARDER: // Can get very chatty, degrade gracefully by being more eager to drop this
+            prio = meshtastic_Config_ReplayConfig_ReplayPriority_LOW;
+            break;
+        case meshtastic_PortNum_REMOTE_HARDWARE_APP:  // Some user likely cares about this, especially if it's for HW control
+        case meshtastic_PortNum_ROUTING_APP:          // Routing packets matter, but are secondary to priority user traffic
+        case meshtastic_PortNum_DETECTION_SENSOR_APP: // Event-triggered, it matters but not as much as human-initiated stuff
+        case meshtastic_PortNum_REPLY_APP:            // Likely human-initiated, but it's just a ping
+        case meshtastic_PortNum_PAXCOUNTER_APP:       // Event-triggered, it matters but not as much as human-initiated stuff
+        case meshtastic_PortNum_REPLAY_APP:           // Anything from the replay module that's important is sent zero-hop
+        case meshtastic_PortNum_POWERSTRESS_APP:      // Seems like a temporary thing that a human will care about
+        case meshtastic_PortNum_PRIVATE_APP:          // Something bespoke, leave it alone
+            prio = meshtastic_Config_ReplayConfig_ReplayPriority_NORMAL;
+            break;
+        case meshtastic_PortNum_ADMIN_APP:        // Remote admin is critical
+        case meshtastic_PortNum_TEXT_MESSAGE_APP: // Text messages are important
+        case meshtastic_PortNum_TEXT_MESSAGE_COMPRESSED_APP:
+        case meshtastic_PortNum_WAYPOINT_APP: // Explicitly user-initiated and often important for a group to reliably RX
+        case meshtastic_PortNum_ALERT_APP:
+        case meshtastic_PortNum_KEY_VERIFICATION_APP: // If this breaks, DMs won't work properly
+            prio = meshtastic_Config_ReplayConfig_ReplayPriority_HIGH;
+            break;
+        default:
+            break; // Don't adjust priority for apps not explicitly handled above
+        }
+
+        /**
+         * Apps not specifically given a priority:
+         *   ZPS_APP - I don't know enough about it to have an opinion
+         *   SIMULATOR_APP - I don't know enough about it to have an opinion
+         *   MAP_REPORT_APP - I don't know enough about it to have an opinion
+         *
+         */
+
+        if (prio > meshtastic_Config_ReplayConfig_ReplayPriority_BACKGROUND && p->to != NODENUM_BROADCAST &&
+            (p->decoded.want_response || p->want_ack))
+            prio = meshtastic_Config_ReplayConfig_ReplayPriority_HIGH; // Unicast messages that want a response or ack are
+                                                                       // important unless from a backgrounded app
+    } else {
+        // Adjustments for packets we can't decrypt
+        if (p->to != NODENUM_BROADCAST) {
+            if (p->want_ack)
+                // ACK-wanted unicast packets are clearly more important
+                prio = meshtastic_Config_ReplayConfig_ReplayPriority_HIGH;
+            else
+                // Assume that unicast stuff has a human who wants it to arrive
+                prio = meshtastic_Config_ReplayConfig_ReplayPriority_NORMAL;
+        } else {
+            if (p->want_ack)
+                // ACK-wanted broadcasts likely matter, but are untargeted
+                prio = meshtastic_Config_ReplayConfig_ReplayPriority_NORMAL;
+            else
+                // Broadcast and no ACK is fire-and-forget
+                prio = meshtastic_Config_ReplayConfig_ReplayPriority_LOW;
+        }
+    }
+
+    return prio;
+}
+
+/**
+ * Learn about a packet if we don't already have an entry for it, and return the entry
+ */
+ReplayPacket *ReplayModule::learnPacket(PacketHash h, meshtastic_Config_ReplayConfig_ReplayPriority prio)
+{
+    ReplayPacket *r = htFind(h);
+    if (r)
+        return r;
+
+    off_t idx = next_idx++ & REPLAY_ENTRY_MASK;
+    recycleSlot(idx);
+    r = &buffer[idx];
+    r->hash = h;
+    r->idx = idx;
+    r->missed = true;
+    r->priority = prio;
+    htInsert(r);
+
+    return r;
+}
+
+/**
+ * Update the priority of a packet we have learned about
+ */
+void ReplayModule::updatePriority(const meshtastic_MeshPacket *p)
+{
+    ReplayPacket *r = htFind(p);
+    if (r)
+        r->priority = getPriority(p);
+}
+
+/**
+ * Get a list of dirty entries pending advertisement
+ */
+size_t ReplayModule::getDirty(ReplayPacket *buf, size_t buf_max, bool fast, bool clear)
+{
+    ReplayPacket *pos = buf;
+    for (off_t i = meshtastic_Config_ReplayConfig_ReplayPriority_HIGH; i >= config.replay.min_advert_priority; i--) {
+        for (off_t j = next_idx + REPLAY_ENTRY_MASK; j >= next_idx && static_cast<size_t>(pos - buf) < buf_max; j--) {
+            uint8_t idx = j & REPLAY_ENTRY_MASK;
+            if (fast) {
+                if (dirty_fast.test(idx)) {
+                    *pos++ = buffer[idx];
+                    if (clear)
+                        dirty_fast.reset(idx);
+                }
+            } else {
+                if (dirty.test(idx)) {
+                    *pos++ = buffer[idx];
+                    if (clear) {
+                        dirty.reset(idx);
+                        dirty_fast.reset(idx);
+                    }
+                }
+            }
+        }
+    }
+    return pos - buf;
+}
+
+/**
+ * Find a packet in the hash table
+ */
+ReplayPacket *ReplayModule::htFind(PacketHash hash)
+{
+    off_t bucket = REPLAY_BUCKET(hash);
+    for (ReplayPacket *r = ht_buckets[bucket]; r; r = r->next) {
+        if (r->hash == hash)
+            return r;
+    }
+    return NULL;
+}
+
+/**
+ * Insert a packet into the hash table
+ */
+void ReplayModule::htInsert(ReplayPacket *r)
+{
+    off_t bucket = REPLAY_BUCKET(r->hash);
+    r->next = ht_buckets[bucket];
+    ht_buckets[bucket] = r;
+}
+
+/**
+ * Remove a packet from the hash table
+ */
+void ReplayModule::htRemove(ReplayPacket *r)
+{
+    off_t bucket = REPLAY_BUCKET(r->hash);
+    for (ReplayPacket **target = &ht_buckets[bucket]; *target; target = &(*target)->next) {
+        if (*target == r) {
+            *target = r->next;
+            r->next = NULL;
+            return;
+        }
+    }
+}
+
+/**
+ * Handle thread notifications
+ */
+void ReplayModule::onNotify(uint32_t notification)
+{
+    switch ((ReplayNotification)notification) {
+    case ReplayNotification::INIT:
+        // Nothing to do yet
+        break;
+    case ReplayNotification::TICK:
+        // Periodic tasks
+        break;
+    default:
+        LOG_WARN("ReplayModule: unknown notification %u", notification);
+        break;
+    }
+}
+
+/**
+ * Prune the packet cache if necessary
+ */
+void ReplayModule::pruneCache()
+{
+    if (cache_bytes <= REPLAY_CACHE_RESERVE)
+        return; // Cache is already smaller than the reserve
+    size_t heap_free_bytes = memGet.getFreeHeap();
+    if (heap_free_bytes >= REPLAY_PRUNE_HEAP_BYTES)
+        return; // We have enough free heap already
+    size_t heap_total_bytes = memGet.getHeapSize();
+    float heap_free_pct = (float)heap_free_bytes / (float)heap_total_bytes;
+    if (heap_free_pct >= REPLAY_PRUNE_HEAP_PCT)
+        return; // We have enough free heap already
+    size_t target_bytes = REPLAY_PRUNE_TARGET;
+    if (cache_bytes - REPLAY_CACHE_RESERVE < target_bytes)
+        target_bytes = cache_bytes - REPLAY_CACHE_RESERVE;
+    unsigned int pruned_packets = 0;
+    unsigned long pruned_bytes = 0;
+    for (off_t i = next_idx; i <= next_idx + REPLAY_ENTRY_MASK && pruned_bytes < target_bytes; i++) {
+        uint8_t idx = i & REPLAY_ENTRY_MASK;
+        ReplayPacket *entry = &buffer[idx];
+        if (entry->cache) {
+            pruned_bytes += sizeof(*entry->cache) + entry->cache->payload_len;
+            cache_bytes -= sizeof(*entry->cache) + entry->cache->payload_len;
+            cache_packets--;
+            packetCache.release(entry->cache);
+            entry->cache = NULL;
+            dirty.reset(idx);
+            dirty_fast.reset(idx);
+            pruned_packets++;
+            if (pruned_bytes >= target_bytes)
+                break;
+        }
+    }
+    routingStats->logEvent(RoutingEvent::REPLAY_CACHE_PACKETS, NULL, cache_packets);
+    routingStats->logEvent(RoutingEvent::REPLAY_CACHE_BYTES, NULL, cache_bytes);
+}
+
+/**
+ * Recycle a slot in the replay buffer
+ */
+void ReplayModule::recycleSlot(uint8_t idx)
+{
+    idx &= REPLAY_ENTRY_MASK;
+    ReplayPacket *entry = &buffer[idx];
+    dirty.reset(idx);
+    dirty_fast.reset(idx);
+    if (entry->cache) {
+        cache_packets--;
+        cache_bytes -= sizeof(*entry->cache) + entry->cache->payload_len;
+        packetCache.release(entry->cache);
+    }
+    *entry = {};
+}