Next hop reliability

src/Mesh.cpp

@@ -9,6 +9,7 @@ void Mesh::begin() {
 
 void Mesh::loop() {
   Dispatcher::loop();
+  processNextHopRetries();
 }
 
 bool Mesh::allowPacketForward(const mesh::Packet* packet) { 
@@ -30,6 +31,12 @@ uint32_t Mesh::getCADFailRetryDelay() const {
   return _rng->nextInt(1, 4)*120;
 }
 
+uint32_t Mesh::getNextHopConfirmTimeout(const Packet* packet) const {
+  // allow for the next hop's own (possibly randomised) forwarding delay, its airtime to repeat, plus margin
+  uint32_t airtime = _radio->getEstAirtimeFor(packet->getRawLength());
+  return airtime * 3 + 2000;
+}
+
 int Mesh::searchPeersByHash(const uint8_t* hash) {
   return 0;  // not found
 }
@@ -39,6 +46,11 @@ int Mesh::searchChannelsByHash(const uint8_t* hash, GroupChannel channels[], int
 }
 
 DispatcherAction Mesh::onRecvPacket(Packet* pkt) {
+  if (pkt->isRouteDirect()) {
+    // any overheard direct packet may be the next hop repeating one of ours -- check before anything else
+    checkNextHopConfirm(pkt);
+  }
+
   if (pkt->isRouteDirect() && pkt->getPayloadType() == PAYLOAD_TYPE_TRACE) {
     if (pkt->path_len < MAX_PATH_SIZE) {
       uint8_t i = 0;
@@ -102,6 +114,17 @@ DispatcherAction Mesh::onRecvPacket(Packet* pkt) {
         _tables->markSeen(pkt);
         removeSelfFromPath(pkt);
 
+        if (pkt->getPathHashCount() > 0) {   // only worth tracking if there is a further hop to overhear
+          registerNextHopConfirm(pkt);
+        } else if (pkt->getPayloadType() == PAYLOAD_TYPE_REQ) {
+          // last hop before the destination -- can't overhear a repeat, but a REQ provokes a
+          // correlated RESPONSE routed back through us, so use that as confirmation instead
+          registerLastHopReplyConfirm(pkt);
+        } else if (pkt->getPayloadType() == PAYLOAD_TYPE_TXT_MSG) {
+          // same, but a TXT_MSG provokes an ACK instead (no id to correlate exactly)
+          registerLastHopAckConfirm(pkt);
+        }
+
         uint32_t d = getDirectRetransmitDelay(pkt);
         return ACTION_RETRANSMIT_DELAYED(0, d);  // Routed traffic is HIGHEST priority 
       }
@@ -341,6 +364,119 @@ void Mesh::removeSelfFromPath(Packet* pkt) {
   }
 }
 
+void Mesh::registerNextHopConfirm(const Packet* pkt) {
+  if (!getNextHopReliabilityEnabled() || getNextHopMaxRetries() == 0) return;
+
+  for (int i = 0; i < MAX_PENDING_NEXTHOP_CONFIRMS; i++) {
+    auto& e = _pending_confirms[i];
+    if (!e.active) {
+      e.active = true;
+      e.kind = NEXTHOP_CONFIRM_REPEAT;
+      e.retries = 0;
+      e.pkt = *pkt;   // keep a copy, so it can be resent unchanged if not confirmed
+      pkt->calculatePacketHash(e.hash);
+      e.deadline = futureMillis(getNextHopConfirmTimeout(pkt));
+      return;
+    }
+  }
+  MESH_DEBUG_PRINTLN("%s Mesh::registerNextHopConfirm(): pending table full, skipping reliability tracking", getLogDateTime());
+}
+
+void Mesh::registerLastHopReplyConfirm(const Packet* pkt) {
+  if (!getNextHopReliabilityEnabled() || getNextHopMaxRetries() == 0) return;
+  if (pkt->payload_len < 2) return;   // need at least [dest_hash, src_hash] to correlate a reply
+
+  for (int i = 0; i < MAX_PENDING_NEXTHOP_CONFIRMS; i++) {
+    auto& e = _pending_confirms[i];
+    if (!e.active) {
+      e.active = true;
+      e.kind = NEXTHOP_CONFIRM_REPLY;
+      e.retries = 0;
+      e.pkt = *pkt;   // keep a copy, so the REQ can be resent unchanged if not confirmed
+      e.expect_dest_hash = pkt->payload[1];   // this REQ's src_hash -> dest_hash we expect on the RESPONSE
+      e.deadline = futureMillis(getNextHopConfirmTimeout(pkt));
+      return;
+    }
+  }
+  MESH_DEBUG_PRINTLN("%s Mesh::registerLastHopReplyConfirm(): pending table full, skipping reliability tracking", getLogDateTime());
+}
+
+void Mesh::registerLastHopAckConfirm(const Packet* pkt) {
+  if (!getNextHopReliabilityEnabled() || getNextHopMaxRetries() == 0) return;
+
+  for (int i = 0; i < MAX_PENDING_NEXTHOP_CONFIRMS; i++) {
+    auto& e = _pending_confirms[i];
+    if (!e.active) {
+      e.active = true;
+      e.kind = NEXTHOP_CONFIRM_ACK_SEEN;
+      e.retries = 0;
+      e.pkt = *pkt;   // keep a copy, so the TXT_MSG can be resent unchanged if not confirmed
+      e.deadline = futureMillis(getNextHopConfirmTimeout(pkt));
+      return;
+    }
+  }
+  MESH_DEBUG_PRINTLN("%s Mesh::registerLastHopAckConfirm(): pending table full, skipping reliability tracking", getLogDateTime());
+}
+
+void Mesh::checkNextHopConfirm(const Packet* pkt) {
+  uint8_t hash[MAX_HASH_SIZE];
+  bool calculated = false;
+  bool is_response = (pkt->getPayloadType() == PAYLOAD_TYPE_RESPONSE && pkt->payload_len > 0);
+  bool is_ack = (pkt->getPayloadType() == PAYLOAD_TYPE_ACK);
+  bool ack_consumed = false;   // only let one overheard ACK confirm one pending entry
+
+  for (int i = 0; i < MAX_PENDING_NEXTHOP_CONFIRMS; i++) {
+    auto& e = _pending_confirms[i];
+    if (!e.active) continue;
+
+    if (e.kind == NEXTHOP_CONFIRM_REPLY) {
+      if (is_response && pkt->payload[0] == e.expect_dest_hash) {
+        e.active = false;   // correlated RESPONSE seen -- confirmed, no retry needed
+      }
+      continue;
+    }
+
+    if (e.kind == NEXTHOP_CONFIRM_ACK_SEEN) {
+      if (is_ack && !ack_consumed) {
+        e.active = false;   // approximate: some direct ACK was overheard -- treat as confirmed
+        ack_consumed = true;
+      }
+      continue;
+    }
+
+    if (!calculated) {
+      pkt->calculatePacketHash(hash);
+      calculated = true;
+    }
+    if (memcmp(hash, e.hash, MAX_HASH_SIZE) == 0) {
+      e.active = false;   // next hop has repeated it -- confirmed, no retry needed
+    }
+  }
+}
+
+void Mesh::processNextHopRetries() {
+  for (int i = 0; i < MAX_PENDING_NEXTHOP_CONFIRMS; i++) {
+    auto& e = _pending_confirms[i];
+    if (!e.active || !millisHasNowPassed(e.deadline)) continue;
+
+    if (e.retries >= getNextHopMaxRetries()) {
+      MESH_DEBUG_PRINTLN("%s Mesh::processNextHopRetries(): giving up, no confirm heard after %d retries", getLogDateTime(), (uint32_t)e.retries);
+      e.active = false;
+      continue;
+    }
+
+    Packet* retry_pkt = obtainNewPacket();
+    if (retry_pkt == NULL) {
+      e.deadline = futureMillis(100);   // packet pool busy, back off briefly and try again
+      continue;
+    }
+    *retry_pkt = e.pkt;
+    e.retries++;
+    e.deadline = futureMillis(getNextHopConfirmTimeout(&e.pkt));
+    sendPacket(retry_pkt, 0);   // resend immediately, same priority as a fresh direct forward
+  }
+}
+
 DispatcherAction Mesh::routeRecvPacket(Packet* packet) {
   uint8_t n = packet->getPathHashCount();
   if (packet->isRouteFlood() && !packet->isMarkedDoNotRetransmit()
@@ -709,6 +845,16 @@ void Mesh::sendDirect(Packet* packet, const uint8_t* path, uint8_t path_len, uin
     } else {
       pri = 0;
     }
+
+    if (packet->getPathHashCount() > 0) {   // there's a next hop to listen for repeating this
+      registerNextHopConfirm(packet);
+    } else if (packet->getPayloadType() == PAYLOAD_TYPE_REQ) {
+      // no next hop to overhear (zero-hop direct REQ) -- wait for the correlated RESPONSE instead
+      registerLastHopReplyConfirm(packet);
+    } else if (packet->getPayloadType() == PAYLOAD_TYPE_TXT_MSG) {
+      // same, but a TXT_MSG provokes an ACK instead (no id to correlate exactly)
+      registerLastHopAckConfirm(packet);
+    }
   }
   _tables->markSeen(packet); // mark this packet as already sent in case it is rebroadcast back to us
   sendPacket(packet, pri, delay_millis);

src/Mesh.h

@@ -20,6 +20,32 @@ class MeshTables {
   virtual void clear(const Packet* packet) = 0;    // remove this packet hash from table
 };
 
+#ifndef MAX_PENDING_NEXTHOP_CONFIRMS
+  #define MAX_PENDING_NEXTHOP_CONFIRMS  4   // max concurrent direct packets awaiting next-hop confirmation
+#endif
+
+#define NEXTHOP_CONFIRM_REPEAT    0   // confirmed by overhearing the next hop repeat this same packet
+#define NEXTHOP_CONFIRM_REPLY     1   // confirmed by seeing a correlated RESPONSE routed back through us
+#define NEXTHOP_CONFIRM_ACK_SEEN  2   // confirmed (heuristically) by overhearing any direct ACK routed back through us
+
+/**
+ * \brief  Tracks a direct (path-routed) packet this node has repeated (or originated), while it
+ *     waits for implicit confirmation of receipt -- either by overhearing the next hop repeat it
+ *     (NEXTHOP_CONFIRM_REPEAT), or, when this is the last hop before the destination and there's
+ *     no next hop to overhear, by seeing the correlated RESPONSE it provokes routed back through
+ *     us (NEXTHOP_CONFIRM_REPLY), or, for a TXT_MSG last hop, by overhearing any direct ACK routed
+ *     back through us (NEXTHOP_CONFIRM_ACK_SEEN -- approximate, since ACKs carry no correlatable id).
+*/
+struct PendingNextHopConfirm {
+  bool active;
+  uint8_t kind;           // NEXTHOP_CONFIRM_REPEAT, NEXTHOP_CONFIRM_REPLY or NEXTHOP_CONFIRM_ACK_SEEN
+  uint8_t retries;
+  uint32_t deadline;      // millis() at which to retry (or give up if retries exhausted)
+  uint8_t hash[MAX_HASH_SIZE];   // kind == NEXTHOP_CONFIRM_REPEAT: hash of the packet we're waiting to hear repeated
+  uint8_t expect_dest_hash;      // kind == NEXTHOP_CONFIRM_REPLY: dest_hash expected on the correlated RESPONSE
+  Packet pkt;             // copy of the packet as it was (re)transmitted, for resending
+};
+
 /**
  * \brief  The next layer in the basic Dispatcher task, Mesh recognises the particular Payload TYPES,
  *     and provides virtual methods for sub-classes on handling incoming, and also preparing outbound Packets.
@@ -28,12 +54,45 @@ class Mesh : public Dispatcher {
   RTCClock* _rtc;
   RNG* _rng;
   MeshTables* _tables;
+  PendingNextHopConfirm _pending_confirms[MAX_PENDING_NEXTHOP_CONFIRMS];
 
   void removeSelfFromPath(Packet* packet);
   void routeDirectRecvAcks(Packet* packet, uint32_t delay_millis);
   //void routeRecvAcks(Packet* packet, uint32_t delay_millis);
   DispatcherAction forwardMultipartDirect(Packet* pkt);
 
+  /**
+   * \brief  Start tracking 'pkt' (just repeated by this node) until the next hop is heard repeating it.
+   */
+  void registerNextHopConfirm(const Packet* pkt);
+
+  /**
+   * \brief  Start tracking 'pkt' (a REQ just delivered to its final destination, with no further
+   *     hop to overhear) until a correlated RESPONSE is seen routed back through this node.
+   */
+  void registerLastHopReplyConfirm(const Packet* pkt);
+
+  /**
+   * \brief  Start tracking 'pkt' (a TXT_MSG just delivered to its final destination, with no
+   *     further hop to overhear) until any direct ACK is seen routed back through this node.
+   *     Approximate: ACKs carry no id to correlate against a specific message.
+   */
+  void registerLastHopAckConfirm(const Packet* pkt);
+
+  /**
+   * \brief  Check an incoming direct packet against the pending-confirm table, and mark any
+   *     match as confirmed -- either a next hop repeating a tracked packet, a RESPONSE
+   *     correlated (by dest_hash) to a tracked last-hop REQ delivery, or any ACK seen following
+   *     a tracked last-hop TXT_MSG delivery.
+   */
+  void checkNextHopConfirm(const Packet* pkt);
+
+  /**
+   * \brief  Called each loop(), resends any pending packets whose confirm deadline has passed,
+   *     up to getNextHopMaxRetries() times, then drops them.
+   */
+  void processNextHopRetries();
+
 protected:
   DispatcherAction onRecvPacket(Packet* pkt) override;
 
@@ -71,6 +130,22 @@ class Mesh : public Dispatcher {
    */
   virtual uint8_t getExtraAckTransmitCount() const;
 
+  /**
+   * \returns  true if 'next-hop reliability' (listen-for-repeat retry) is enabled for repeated
+   *     direct packets. Default is enabled wherever allowPacketForward() also permits forwarding.
+   */
+  virtual bool getNextHopReliabilityEnabled() const { return true; }
+
+  /**
+   * \returns  max number of retries (resends) attempted, if no repeat from the next hop is heard.
+   */
+  virtual uint8_t getNextHopMaxRetries() const { return 3; }
+
+  /**
+   * \returns  number of milliseconds to wait for the next hop to repeat 'packet', before retrying.
+   */
+  virtual uint32_t getNextHopConfirmTimeout(const Packet* packet) const;
+
   /**
    * \brief  Perform search of local DB of peers/contacts.
    * \returns  Number of peers with matching hash
@@ -169,6 +244,7 @@ class Mesh : public Dispatcher {
   Mesh(Radio& radio, MillisecondClock& ms, RNG& rng, RTCClock& rtc, PacketManager& mgr, MeshTables& tables)
     : Dispatcher(radio, ms, mgr), _rng(&rng), _rtc(&rtc), _tables(&tables)
   {
+    memset(_pending_confirms, 0, sizeof(_pending_confirms));
   }
 
   MeshTables* getTables() const { return _tables; }