ack.hpp

Go to the documentation of this file.

#ifndef PAINLESS_MESH_ACK_HPP
#define PAINLESS_MESH_ACK_HPP
 
#include <functional>
#include <list>
#include <map>
#include <set>
 
#include "painlessmesh/configuration.hpp"
#include "painlessmesh/logger.hpp"
#include "painlessmesh/plugin.hpp"
 
extern painlessmesh::logger::LogClass Log;
 
// Maximum number of messages that may await acknowledgment at once.
// Bounded to protect the small ESP8266 heap; override at build time if
// your application legitimately needs more concurrent tracked sends.
#ifndef PAINLESSMESH_MAX_PENDING_ACKS
#define PAINLESSMESH_MAX_PENDING_ACKS 32
#endif
 
// Maximum number of delayed broadcast acknowledgments queued by a receiver.
// A single shared scheduler task drains the queue after the jitter delay.
#ifndef PAINLESSMESH_MAX_QUEUED_BROADCAST_ACKS
#define PAINLESSMESH_MAX_QUEUED_BROADCAST_ACKS 32
#endif
 
namespace painlessmesh {
 
namespace ack {
 
typedef std::function<void(uint32_t nodeId, bool delivered, uint32_t latencyMs)>
    deliveryCallback_t;
 
class MessageAckPackage : public plugin::SinglePackage {
 public:
  uint32_t messageId = 0;
 
  MessageAckPackage() : SinglePackage(protocol::MESSAGE_ACK) {}
 
  MessageAckPackage(uint32_t fromNode, uint32_t destNode, uint32_t msgId)
      : SinglePackage(protocol::MESSAGE_ACK) {
    from = fromNode;
    dest = destNode;
    messageId = msgId;
  }
 
  MessageAckPackage(JsonObject jsonObj) : SinglePackage(jsonObj) {
    messageId = jsonObj["msgId"] | (uint32_t)0;
  }
 
  JsonObject addTo(JsonObject&& jsonObj) const {
    jsonObj = SinglePackage::addTo(std::move(jsonObj));
    jsonObj["msgId"] = messageId;
    return jsonObj;
  }
 
#if ARDUINOJSON_VERSION_MAJOR < 7
  size_t jsonObjectSize() const { return JSON_OBJECT_SIZE(noJsonFields + 1); }
#endif
};
 
struct PendingAck {
  deliveryCallback_t callback;
  uint32_t sentAt = 0;
  uint32_t timeoutMs = 0;
  std::set<uint32_t> waitingFor;
};
 
struct DeliveryResult {
  deliveryCallback_t callback;
  uint32_t nodeId = 0;
  bool delivered = false;
  uint32_t latencyMs = 0;
 
  DeliveryResult(deliveryCallback_t callback, uint32_t nodeId, bool delivered,
                 uint32_t latencyMs)
      : callback(callback),
        nodeId(nodeId),
        delivered(delivered),
        latencyMs(latencyMs) {}
};
 
class AckTracker {
 public:
  void seed(uint32_t value) { counter = value; }
 
  uint32_t nextMessageId() {
    if (++counter == 0) ++counter;
    return counter;
  }
 
  bool full() const { return entries.size() >= PAINLESSMESH_MAX_PENDING_ACKS; }
 
  bool track(uint32_t msgId, const std::list<uint32_t>& destinations,
             deliveryCallback_t callback, uint32_t timeoutMs, uint32_t now) {
    if (!callback || destinations.empty()) return false;
    if (full()) {
      Log(logger::ERROR,
          "AckTracker: pending-ack limit (%u) reached, not tracking %u\n",
          (unsigned)PAINLESSMESH_MAX_PENDING_ACKS, msgId);
      return false;
    }
    PendingAck entry;
    entry.callback = callback;
    entry.sentAt = now;
    entry.timeoutMs = timeoutMs;
    entry.waitingFor.insert(destinations.begin(), destinations.end());
    entries[msgId] = std::move(entry);
    return true;
  }
 
  bool handleAck(uint32_t msgId, uint32_t fromNode, uint32_t now) {
    std::list<DeliveryResult> results;
    auto matched = collectAck(msgId, fromNode, now, results);
    for (auto&& result : results) {
      result.callback(result.nodeId, result.delivered, result.latencyMs);
    }
    return matched;
  }
 
  bool collectAck(uint32_t msgId, uint32_t fromNode, uint32_t now,
                  std::list<DeliveryResult>& results) {
    auto it = entries.find(msgId);
    if (it == entries.end()) return false;
    if (it->second.waitingFor.count(fromNode) == 0) return false;
    auto latency = (uint32_t)(now - it->second.sentAt);
    if (latency >= it->second.timeoutMs) {
      // The polling task may not have run exactly at the deadline. Reject
      // this late ACK and expire every destination still waiting on the
      // same message before invoking user code (callbacks may reenter us).
      auto expired = std::move(it->second);
      entries.erase(it);
      for (auto&& nodeId : expired.waitingFor) {
        Log(logger::COMMUNICATION,
            "AckTracker: timeout waiting for ack from %u\n", nodeId);
        results.push_back(
            DeliveryResult(expired.callback, nodeId, false, expired.timeoutMs));
      }
      return false;
    }
    it->second.waitingFor.erase(fromNode);
    auto callback = it->second.callback;
    if (it->second.waitingFor.empty()) entries.erase(it);
    results.push_back(DeliveryResult(callback, fromNode, true, latency));
    return true;
  }
 
  size_t expire(uint32_t now) {
    std::list<DeliveryResult> results;
    collectExpired(now, results);
    for (const auto& result : results) {
      result.callback(result.nodeId, result.delivered, result.latencyMs);
    }
    return entries.size();
  }
 
  size_t collectExpired(uint32_t now, std::list<DeliveryResult>& results) {
    // Collect expired entries and erase them from the map BEFORE exposing
    // results to user callbacks: a callback may reenter this tracker (track a
    // retry, call clear() via mesh.stop(), or poll checkAcks()), which
    // would invalidate a live iterator into `entries`.
    std::list<PendingAck> expired;
    for (auto it = entries.begin(); it != entries.end();) {
      if ((uint32_t)(now - it->second.sentAt) >= it->second.timeoutMs) {
        expired.push_back(std::move(it->second));
        it = entries.erase(it);
      } else {
        ++it;
      }
    }
    for (auto&& entry : expired) {
      for (auto&& nodeId : entry.waitingFor) {
        Log(logger::COMMUNICATION,
            "AckTracker: timeout waiting for ack from %u\n", nodeId);
        results.push_back(
            DeliveryResult(entry.callback, nodeId, false, entry.timeoutMs));
      }
    }
    return entries.size();
  }
 
  size_t pending() const { return entries.size(); }
 
  void clear() { entries.clear(); }
 
 protected:
  std::map<uint32_t, PendingAck> entries;
  uint32_t counter = 0;
};
 
}  // namespace ack
}  // namespace painlessmesh
 
#endif